SHA256
Документировать API и сервис агента-кодера
This commit is contained in:
AidarKC
@@ -0,0 +1,4 @@
|
||||
package shine.agent.botcoder.queue;
|
||||
|
||||
public record FailureResult(boolean willRetry, int attempts, int maxRetries) {
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package shine.agent.botcoder.queue;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.UUID;
|
||||
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
public class QueueJob {
|
||||
public String id;
|
||||
public QueueStatus status;
|
||||
public String type;
|
||||
public long chatId;
|
||||
public int messageId;
|
||||
public String username;
|
||||
public String text;
|
||||
public String telegramFileId;
|
||||
public String historyFile;
|
||||
public String createdAt;
|
||||
public String updatedAt;
|
||||
public String activeSince;
|
||||
public int attempts;
|
||||
public String retryReason;
|
||||
public String lastError;
|
||||
|
||||
public static QueueJob textJob(long chatId, int messageId, String username, String text, String historyFile) {
|
||||
QueueJob job = baseJob(chatId, messageId, username, historyFile);
|
||||
job.type = "text";
|
||||
job.text = text;
|
||||
return job;
|
||||
}
|
||||
|
||||
public static QueueJob voiceJob(long chatId, int messageId, String username, String fileId, String historyFile) {
|
||||
QueueJob job = baseJob(chatId, messageId, username, historyFile);
|
||||
job.type = "voice";
|
||||
job.telegramFileId = fileId;
|
||||
return job;
|
||||
}
|
||||
|
||||
private static QueueJob baseJob(long chatId, int messageId, String username, String historyFile) {
|
||||
QueueJob job = new QueueJob();
|
||||
String now = Instant.now().toString();
|
||||
job.id = UUID.randomUUID().toString();
|
||||
job.status = QueueStatus.PENDING;
|
||||
job.chatId = chatId;
|
||||
job.messageId = messageId;
|
||||
job.username = username;
|
||||
job.historyFile = historyFile;
|
||||
job.createdAt = now;
|
||||
job.updatedAt = now;
|
||||
job.attempts = 0;
|
||||
return job;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
package shine.agent.botcoder.queue;
|
||||
|
||||
public enum QueueStatus {
|
||||
PENDING,
|
||||
ACTIVE
|
||||
}
|
||||
@@ -0,0 +1,203 @@
|
||||
package shine.agent.botcoder.queue;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import shine.agent.botcoder.state.RuntimeStateStore;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.StandardOpenOption;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
public class QueueStore {
|
||||
|
||||
private final Path queueFile;
|
||||
private final RuntimeStateStore stateStore;
|
||||
private final ObjectMapper mapper;
|
||||
private final List<QueueJob> jobs;
|
||||
|
||||
public QueueStore(Path queueFile, RuntimeStateStore stateStore) throws IOException {
|
||||
this.queueFile = queueFile;
|
||||
this.stateStore = stateStore;
|
||||
this.mapper = new ObjectMapper();
|
||||
Path parent = queueFile.getParent();
|
||||
if (parent != null) {
|
||||
Files.createDirectories(parent);
|
||||
}
|
||||
this.jobs = loadQueue();
|
||||
persistQueue();
|
||||
}
|
||||
|
||||
public synchronized void enqueue(QueueJob job) throws IOException {
|
||||
jobs.add(job);
|
||||
persistQueue();
|
||||
}
|
||||
|
||||
public synchronized List<String> recoverActiveJobs() throws IOException {
|
||||
List<String> recovered = new ArrayList<>();
|
||||
String now = Instant.now().toString();
|
||||
for (QueueJob job : jobs) {
|
||||
if (job.status == QueueStatus.ACTIVE) {
|
||||
job.status = QueueStatus.PENDING;
|
||||
job.retryReason = "service_restart_recovery";
|
||||
job.updatedAt = now;
|
||||
recovered.add(job.id);
|
||||
}
|
||||
}
|
||||
stateStore.setActiveJobId(null);
|
||||
persistQueue();
|
||||
return recovered;
|
||||
}
|
||||
|
||||
public synchronized Optional<QueueJob> activateNext() throws IOException {
|
||||
for (QueueJob job : jobs) {
|
||||
if (job.status == QueueStatus.PENDING) {
|
||||
job.status = QueueStatus.ACTIVE;
|
||||
job.activeSince = Instant.now().toString();
|
||||
job.updatedAt = job.activeSince;
|
||||
stateStore.setActiveJobId(job.id);
|
||||
persistQueue();
|
||||
return Optional.of(job);
|
||||
}
|
||||
}
|
||||
return Optional.empty();
|
||||
}
|
||||
|
||||
public synchronized void markDone(String jobId) throws IOException {
|
||||
Iterator<QueueJob> iterator = jobs.iterator();
|
||||
while (iterator.hasNext()) {
|
||||
QueueJob job = iterator.next();
|
||||
if (job.id.equals(jobId)) {
|
||||
iterator.remove();
|
||||
break;
|
||||
}
|
||||
}
|
||||
stateStore.setActiveJobId(null);
|
||||
persistQueue();
|
||||
}
|
||||
|
||||
public synchronized Optional<QueueJob> getActiveJob() {
|
||||
return jobs.stream().filter(j -> j.status == QueueStatus.ACTIVE).findFirst();
|
||||
}
|
||||
|
||||
public synchronized int pendingCount() {
|
||||
int count = 0;
|
||||
for (QueueJob job : jobs) {
|
||||
if (job.status == QueueStatus.PENDING) {
|
||||
count++;
|
||||
}
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
public synchronized int totalCount() {
|
||||
return jobs.size();
|
||||
}
|
||||
|
||||
public synchronized List<QueueJob> snapshot() {
|
||||
return new ArrayList<>(jobs);
|
||||
}
|
||||
|
||||
public synchronized boolean cancelActiveJob(String reason) throws IOException {
|
||||
for (Iterator<QueueJob> iterator = jobs.iterator(); iterator.hasNext(); ) {
|
||||
QueueJob job = iterator.next();
|
||||
if (job.status == QueueStatus.ACTIVE) {
|
||||
iterator.remove();
|
||||
stateStore.setActiveJobId(null);
|
||||
persistQueue();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public synchronized int cancelAll(String reason) throws IOException {
|
||||
int size = jobs.size();
|
||||
if (size == 0) {
|
||||
return 0;
|
||||
}
|
||||
jobs.clear();
|
||||
stateStore.setActiveJobId(null);
|
||||
persistQueue();
|
||||
return size;
|
||||
}
|
||||
|
||||
public synchronized boolean cancelByIdPrefix(String idPrefix) throws IOException {
|
||||
if (idPrefix == null || idPrefix.isBlank()) {
|
||||
return false;
|
||||
}
|
||||
String normalized = idPrefix.trim().toLowerCase();
|
||||
for (Iterator<QueueJob> iterator = jobs.iterator(); iterator.hasNext(); ) {
|
||||
QueueJob job = iterator.next();
|
||||
if (job.id != null && job.id.toLowerCase().startsWith(normalized)) {
|
||||
if (job.status == QueueStatus.ACTIVE) {
|
||||
stateStore.setActiveJobId(null);
|
||||
}
|
||||
iterator.remove();
|
||||
persistQueue();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public synchronized FailureResult markFailed(String jobId, String error, int maxRetries) throws IOException {
|
||||
for (Iterator<QueueJob> it = jobs.iterator(); it.hasNext(); ) {
|
||||
QueueJob job = it.next();
|
||||
if (!job.id.equals(jobId)) {
|
||||
continue;
|
||||
}
|
||||
job.attempts = job.attempts + 1;
|
||||
job.lastError = error;
|
||||
job.updatedAt = Instant.now().toString();
|
||||
stateStore.setActiveJobId(null);
|
||||
|
||||
if (job.attempts < maxRetries) {
|
||||
job.status = QueueStatus.PENDING;
|
||||
job.retryReason = error;
|
||||
persistQueue();
|
||||
return new FailureResult(true, job.attempts, maxRetries);
|
||||
}
|
||||
|
||||
it.remove();
|
||||
persistQueue();
|
||||
return new FailureResult(false, job.attempts, maxRetries);
|
||||
}
|
||||
|
||||
stateStore.setActiveJobId(null);
|
||||
persistQueue();
|
||||
return new FailureResult(false, maxRetries, maxRetries);
|
||||
}
|
||||
|
||||
private List<QueueJob> loadQueue() throws IOException {
|
||||
List<QueueJob> loaded = new ArrayList<>();
|
||||
if (!Files.exists(queueFile)) {
|
||||
return loaded;
|
||||
}
|
||||
for (String line : Files.readAllLines(queueFile, StandardCharsets.UTF_8)) {
|
||||
String trimmed = line.trim();
|
||||
if (trimmed.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
loaded.add(mapper.readValue(trimmed, QueueJob.class));
|
||||
}
|
||||
return loaded;
|
||||
}
|
||||
|
||||
private void persistQueue() throws IOException {
|
||||
Files.writeString(queueFile, "", StandardCharsets.UTF_8);
|
||||
for (QueueJob job : jobs) {
|
||||
Files.writeString(
|
||||
queueFile,
|
||||
mapper.writeValueAsString(job) + System.lineSeparator(),
|
||||
StandardCharsets.UTF_8,
|
||||
StandardOpenOption.APPEND
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user