From 4ad3b3928feaf275cee3d9eed1261748a2843715 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:10:28 +0530 Subject: [PATCH 1/8] feat(java): add signed and encrypted serializers --- .../serialization/EncryptedSerializer.java | 61 +++++++++++++++++++ .../serialization/SignedSerializer.java | 57 +++++++++++++++++ 2 files changed, 118 insertions(+) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/serialization/EncryptedSerializer.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/serialization/SignedSerializer.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/serialization/EncryptedSerializer.java b/sdks/java/src/main/java/org/byteveda/taskito/serialization/EncryptedSerializer.java new file mode 100644 index 00000000..15a66b70 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/serialization/EncryptedSerializer.java @@ -0,0 +1,61 @@ +package org.byteveda.taskito.serialization; + +import java.security.SecureRandom; +import java.util.Arrays; +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import org.byteveda.taskito.TaskitoException; + +/** + * Wraps a delegate serializer with AES-GCM encryption. Each payload is prefixed + * with a fresh 12-byte IV; the GCM tag provides authentication. The key must be + * 16, 24, or 32 bytes (AES-128/192/256). + */ +public final class EncryptedSerializer implements Serializer { + private static final String TRANSFORMATION = "AES/GCM/NoPadding"; + private static final int IV_LENGTH = 12; + private static final int TAG_BITS = 128; + + private final Serializer delegate; + private final SecretKeySpec key; + private final SecureRandom random = new SecureRandom(); + + public EncryptedSerializer(Serializer delegate, byte[] key) { + this.delegate = delegate; + this.key = new SecretKeySpec(key, "AES"); + } + + @Override + public byte[] serialize(Object value) { + try { + byte[] iv = new byte[IV_LENGTH]; + random.nextBytes(iv); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + cipher.init(Cipher.ENCRYPT_MODE, key, new GCMParameterSpec(TAG_BITS, iv)); + byte[] ciphertext = cipher.doFinal(delegate.serialize(value)); + byte[] out = new byte[IV_LENGTH + ciphertext.length]; + System.arraycopy(iv, 0, out, 0, IV_LENGTH); + System.arraycopy(ciphertext, 0, out, IV_LENGTH, ciphertext.length); + return out; + } catch (Exception e) { + throw new TaskitoException("encryption failed", e); + } + } + + @Override + public T deserialize(byte[] bytes, Class type) { + if (bytes.length < IV_LENGTH) { + throw new TaskitoException("encrypted payload is too short"); + } + try { + byte[] iv = Arrays.copyOfRange(bytes, 0, IV_LENGTH); + byte[] ciphertext = Arrays.copyOfRange(bytes, IV_LENGTH, bytes.length); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + cipher.init(Cipher.DECRYPT_MODE, key, new GCMParameterSpec(TAG_BITS, iv)); + return delegate.deserialize(cipher.doFinal(ciphertext), type); + } catch (Exception e) { + throw new TaskitoException("decryption failed", e); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/serialization/SignedSerializer.java b/sdks/java/src/main/java/org/byteveda/taskito/serialization/SignedSerializer.java new file mode 100644 index 00000000..b4e42ea6 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/serialization/SignedSerializer.java @@ -0,0 +1,57 @@ +package org.byteveda.taskito.serialization; + +import java.security.MessageDigest; +import java.util.Arrays; +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import org.byteveda.taskito.TaskitoException; + +/** + * Wraps a delegate serializer, prefixing each payload with an HMAC-SHA256 tag. + * Deserialization verifies the tag (constant-time) and rejects tampered bytes. + */ +public final class SignedSerializer implements Serializer { + private static final String ALGORITHM = "HmacSHA256"; + private static final int MAC_LENGTH = 32; + + private final Serializer delegate; + private final byte[] key; + + public SignedSerializer(Serializer delegate, byte[] key) { + this.delegate = delegate; + this.key = key.clone(); + } + + @Override + public byte[] serialize(Object value) { + byte[] body = delegate.serialize(value); + byte[] mac = mac(body); + byte[] out = new byte[MAC_LENGTH + body.length]; + System.arraycopy(mac, 0, out, 0, MAC_LENGTH); + System.arraycopy(body, 0, out, MAC_LENGTH, body.length); + return out; + } + + @Override + public T deserialize(byte[] bytes, Class type) { + if (bytes.length < MAC_LENGTH) { + throw new TaskitoException("signed payload is too short"); + } + byte[] mac = Arrays.copyOfRange(bytes, 0, MAC_LENGTH); + byte[] body = Arrays.copyOfRange(bytes, MAC_LENGTH, bytes.length); + if (!MessageDigest.isEqual(mac, mac(body))) { + throw new TaskitoException("signature mismatch"); + } + return delegate.deserialize(body, type); + } + + private byte[] mac(byte[] body) { + try { + Mac mac = Mac.getInstance(ALGORITHM); + mac.init(new SecretKeySpec(key, ALGORITHM)); + return mac.doFinal(body); + } catch (Exception e) { + throw new TaskitoException("HMAC computation failed", e); + } + } +} From e9fc48b44690507b18db0c26e783f7103dce57f1 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:10:28 +0530 Subject: [PATCH 2/8] feat(java): add middleware hooks Queue.use() registers Middleware; onEnqueue runs on the producer, before/after/onError wrap handler execution, and the outcome hooks fire from onOutcome. --- .../org/byteveda/taskito/DefaultQueue.java | 25 +++++++-- .../main/java/org/byteveda/taskito/Queue.java | 4 ++ .../java/org/byteveda/taskito/Worker.java | 11 ++-- .../taskito/WorkerDispatchBridge.java | 53 ++++++++++++++++--- .../taskito/middleware/EnqueueContext.java | 36 +++++++++++++ .../taskito/middleware/Middleware.java | 27 ++++++++++ .../taskito/middleware/TaskContext.java | 12 +++++ .../taskito/middleware/package-info.java | 2 + 8 files changed, 157 insertions(+), 13 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/middleware/Middleware.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/middleware/TaskContext.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/middleware/package-info.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java index 72c54dee..ef396084 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java @@ -7,6 +7,9 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.CopyOnWriteArrayList; +import org.byteveda.taskito.middleware.EnqueueContext; +import org.byteveda.taskito.middleware.Middleware; import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.QueueBackend; @@ -20,12 +23,19 @@ final class DefaultQueue implements Queue { private final QueueBackend backend; private final Serializer serializer; + private final List middleware = new CopyOnWriteArrayList<>(); DefaultQueue(QueueBackend backend, Serializer serializer) { this.backend = backend; this.serializer = serializer; } + @Override + public Queue use(Middleware middleware) { + this.middleware.add(middleware); + return this; + } + // ── Producer ──────────────────────────────────────────────────── @Override @@ -35,12 +45,21 @@ public String enqueue(Task task, T payload) { @Override public String enqueue(Task task, T payload, EnqueueOptions options) { - return backend.enqueue(task.name(), serializer.serialize(payload), encode(options)); + return dispatchEnqueue(task.name(), payload, options); } @Override public String enqueue(String taskName, Object payload) { - return backend.enqueue(taskName, serializer.serialize(payload), encode(EnqueueOptions.none())); + return dispatchEnqueue(taskName, payload, EnqueueOptions.none()); + } + + /** Run onEnqueue middleware, then serialize and submit the (possibly rewritten) job. */ + private String dispatchEnqueue(String taskName, Object payload, EnqueueOptions options) { + EnqueueContext context = new EnqueueContext(taskName, payload, options); + for (Middleware m : middleware) { + m.onEnqueue(context); + } + return backend.enqueue(taskName, serializer.serialize(context.payload()), encode(context.options())); } @Override @@ -212,7 +231,7 @@ public List getTaskLogs(String jobId) { @Override public Worker.Builder worker() { - return Worker.builder(backend, serializer); + return Worker.builder(backend, serializer, middleware); } @Override diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java index dbfab619..058b39e4 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java @@ -3,9 +3,13 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import org.byteveda.taskito.middleware.Middleware; /** A Taskito queue. Obtain one from {@link Taskito#builder()}. */ public interface Queue extends AutoCloseable { + /** Register cross-cutting middleware (enqueue + worker hooks); returns {@code this}. */ + Queue use(Middleware middleware); + // ── Producer ──────────────────────────────────────────────────── /** Enqueue a typed payload using the task's default options; returns the job id. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/Worker.java index d4b52721..b02c6185 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Worker.java @@ -16,6 +16,7 @@ import org.byteveda.taskito.events.Emitter; import org.byteveda.taskito.events.EventName; import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.middleware.Middleware; import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.QueueBackend; import org.byteveda.taskito.spi.WorkerControl; @@ -38,8 +39,8 @@ private Worker(WorkerControl control, ExecutorService executor) { this.executor = executor; } - public static Builder builder(QueueBackend backend, Serializer serializer) { - return new Builder(backend, serializer); + public static Builder builder(QueueBackend backend, Serializer serializer, List middleware) { + return new Builder(backend, serializer, middleware); } /** Stop dispatching; in-flight jobs continue to drain. */ @@ -85,6 +86,7 @@ public static final class Builder { private final QueueBackend backend; private final Serializer serializer; + private final List middleware; private final Map handlers = new HashMap<>(); private final Map>> listeners = new EnumMap<>(EventName.class); @@ -93,9 +95,10 @@ public static final class Builder { private Integer channelCapacity; private Integer batchSize; - Builder(QueueBackend backend, Serializer serializer) { + Builder(QueueBackend backend, Serializer serializer, List middleware) { this.backend = backend; this.serializer = serializer; + this.middleware = middleware; } public Builder handle(String taskName, Class payloadType, TaskFunction handler) { @@ -140,7 +143,7 @@ public Worker start() { Emitter emitter = new Emitter(); listeners.forEach((name, bound) -> bound.forEach(listener -> emitter.on(name, listener))); WorkerDispatchBridge bridge = - new WorkerDispatchBridge(handlers, serializer, executor, emitter); + new WorkerDispatchBridge(handlers, serializer, executor, emitter, middleware); WorkerControl control = backend.startWorker(bridge, encodeOptions()); bridge.bind(control); return new Worker(control, executor); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java index 00bc488e..0f5c95ba 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java @@ -1,25 +1,30 @@ package org.byteveda.taskito; +import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import org.byteveda.taskito.events.Emitter; import org.byteveda.taskito.events.EventName; import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.middleware.Middleware; +import org.byteveda.taskito.middleware.TaskContext; import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.WorkerBridge; import org.byteveda.taskito.spi.WorkerControl; /** * Bridges native job dispatch to registered handlers. {@code onJob} hands work to - * an executor and completes it via the {@link WorkerControl}; {@code onOutcome} - * fans finished jobs out to event listeners. + * an executor, runs middleware around the handler, and completes via the + * {@link WorkerControl}; {@code onOutcome} fans finished jobs out to middleware + * and event listeners. */ final class WorkerDispatchBridge implements WorkerBridge { private final Map handlers; private final Serializer serializer; private final ExecutorService executor; private final Emitter emitter; + private final List middleware; // Resolved once startWorker returns; job tasks await it before completing. private final CompletableFuture control = new CompletableFuture<>(); @@ -27,11 +32,13 @@ final class WorkerDispatchBridge implements WorkerBridge { Map handlers, Serializer serializer, ExecutorService executor, - Emitter emitter) { + Emitter emitter, + List middleware) { this.handlers = handlers; this.serializer = serializer; this.executor = executor; this.emitter = emitter; + this.middleware = middleware; } void bind(WorkerControl control) { @@ -40,28 +47,62 @@ void bind(WorkerControl control) { @Override public void onJob(long token, String jobId, String taskName, byte[] payload) { - executor.execute(() -> runJob(token, taskName, payload)); + executor.execute(() -> runJob(token, jobId, taskName, payload)); } - private void runJob(long token, String taskName, byte[] payload) { + private void runJob(long token, String jobId, String taskName, byte[] payload) { WorkerControl bound = control.join(); RegisteredTask task = handlers.get(taskName); if (task == null) { bound.failJob(token, "no handler registered for task '" + taskName + "'"); return; } + TaskContext context = new TaskContext(jobId, taskName); try { + for (Middleware m : middleware) { + m.before(context); + } Object argument = serializer.deserialize(payload, task.payloadType); Object result = task.handler.apply(argument); + for (Middleware m : middleware) { + m.after(context, result); + } bound.completeJob(token, serializer.serialize(result)); } catch (Throwable t) { + for (Middleware m : middleware) { + m.onError(context, t); + } bound.failJob(token, describe(t)); } } @Override public void onOutcome(String kind, String jobId, String taskName, String error, int retryCount, boolean timedOut) { - emitter.emit(new OutcomeEvent(EventName.fromKind(kind), jobId, taskName, error, retryCount, timedOut)); + EventName name = EventName.fromKind(kind); + OutcomeEvent event = new OutcomeEvent(name, jobId, taskName, error, retryCount, timedOut); + emitter.emit(event); + for (Middleware m : middleware) { + dispatch(m, name, event); + } + } + + private static void dispatch(Middleware m, EventName name, OutcomeEvent event) { + switch (name) { + case SUCCESS: + m.onCompleted(event); + break; + case RETRY: + m.onRetry(event); + break; + case DEAD: + m.onDeadLetter(event); + break; + case CANCELLED: + m.onCancel(event); + break; + default: + break; + } } private static String describe(Throwable t) { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java b/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java new file mode 100644 index 00000000..16193ec1 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java @@ -0,0 +1,36 @@ +package org.byteveda.taskito.middleware; + +import org.byteveda.taskito.EnqueueOptions; + +/** + * A job being enqueued, passed to {@link Middleware#onEnqueue} before + * serialization. Replace the payload or options to rewrite the job; throw to + * abort the enqueue. + */ +public final class EnqueueContext { + public final String taskName; + private Object payload; + private EnqueueOptions options; + + public EnqueueContext(String taskName, Object payload, EnqueueOptions options) { + this.taskName = taskName; + this.payload = payload; + this.options = options; + } + + public Object payload() { + return payload; + } + + public void payload(Object payload) { + this.payload = payload; + } + + public EnqueueOptions options() { + return options; + } + + public void options(EnqueueOptions options) { + this.options = options; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/middleware/Middleware.java b/sdks/java/src/main/java/org/byteveda/taskito/middleware/Middleware.java new file mode 100644 index 00000000..f527fa81 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/middleware/Middleware.java @@ -0,0 +1,27 @@ +package org.byteveda.taskito.middleware; + +import org.byteveda.taskito.events.OutcomeEvent; + +/** + * Cross-cutting hooks around enqueue, task execution, and job outcomes. All are + * optional. {@code onEnqueue} runs on the producer before serialization; + * {@code before}/{@code after}/{@code onError} wrap execution; the outcome hooks + * fire after the core decides the result. Register with {@link org.byteveda.taskito.Queue#use}. + */ +public interface Middleware { + default void onEnqueue(EnqueueContext context) {} + + default void before(TaskContext context) {} + + default void after(TaskContext context, Object result) {} + + default void onError(TaskContext context, Throwable error) {} + + default void onCompleted(OutcomeEvent event) {} + + default void onRetry(OutcomeEvent event) {} + + default void onDeadLetter(OutcomeEvent event) {} + + default void onCancel(OutcomeEvent event) {} +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/middleware/TaskContext.java b/sdks/java/src/main/java/org/byteveda/taskito/middleware/TaskContext.java new file mode 100644 index 00000000..43d07879 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/middleware/TaskContext.java @@ -0,0 +1,12 @@ +package org.byteveda.taskito.middleware; + +/** Identifies a task as it executes on a worker. */ +public final class TaskContext { + public final String jobId; + public final String taskName; + + public TaskContext(String jobId, String taskName) { + this.jobId = jobId; + this.taskName = taskName; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/middleware/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/middleware/package-info.java new file mode 100644 index 00000000..10336969 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/middleware/package-info.java @@ -0,0 +1,2 @@ +/** Cross-cutting worker/enqueue hooks: Middleware plus TaskContext and EnqueueContext. */ +package org.byteveda.taskito.middleware; From dd18ac4cd6745fda1d7da53cbb5511736b93f64e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:10:28 +0530 Subject: [PATCH 3/8] feat(java): add dashboard HTTP server JDK httpserver-backed read/action API over a Queue, serving the snake_case + Unix-ms contract with optional token auth and static SPA serving. --- .../byteveda/taskito/dashboard/Contract.java | 88 +++++ .../taskito/dashboard/DashboardServer.java | 339 ++++++++++++++++++ .../taskito/dashboard/package-info.java | 2 + 3 files changed, 429 insertions(+) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/dashboard/package-info.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java new file mode 100644 index 00000000..dee31308 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java @@ -0,0 +1,88 @@ +package org.byteveda.taskito.dashboard; + +import java.util.LinkedHashMap; +import java.util.Map; +import org.byteveda.taskito.DeadJob; +import org.byteveda.taskito.Job; +import org.byteveda.taskito.QueueStats; +import org.byteveda.taskito.TaskMetric; +import org.byteveda.taskito.WorkerInfo; + +/** + * Maps the SDK's camelCase types to the snake_case JSON contract the React SPA + * expects. Timestamps are Unix milliseconds throughout (never scaled). + */ +final class Contract { + private Contract() {} + + static Map stats(QueueStats s) { + Map m = new LinkedHashMap<>(); + m.put("pending", s.pending); + m.put("running", s.running); + m.put("completed", s.completed); + m.put("failed", s.failed); + m.put("dead", s.dead); + m.put("cancelled", s.cancelled); + return m; + } + + static Map job(Job j) { + Map m = new LinkedHashMap<>(); + m.put("id", j.id); + m.put("task_name", j.taskName); + m.put("queue", j.queue); + m.put("status", j.status.wire()); + m.put("priority", j.priority); + m.put("progress", j.progress); + m.put("retry_count", j.retryCount); + m.put("max_retries", j.maxRetries); + m.put("created_at", j.createdAt); + m.put("scheduled_at", j.scheduledAt); + m.put("started_at", j.startedAt); + m.put("completed_at", j.completedAt); + m.put("timeout_ms", j.timeoutMs); + m.put("error", j.error); + m.put("unique_key", j.uniqueKey); + m.put("namespace", j.namespace); + return m; + } + + static Map worker(WorkerInfo w) { + Map m = new LinkedHashMap<>(); + m.put("worker_id", w.workerId); + m.put("queues", w.queues); + m.put("status", w.status); + m.put("last_heartbeat", w.lastHeartbeat); + m.put("registered_at", w.startedAt == null ? w.lastHeartbeat : w.startedAt); + m.put("hostname", w.hostname); + m.put("pid", w.pid); + m.put("pool_type", w.poolType); + m.put("tags", w.tags); + return m; + } + + static Map dead(DeadJob d) { + Map m = new LinkedHashMap<>(); + m.put("id", d.id); + m.put("original_job_id", d.originalJobId); + m.put("task_name", d.taskName); + m.put("queue", d.queue); + m.put("error", d.error); + m.put("retry_count", d.retryCount); + m.put("failed_at", d.failedAt); + m.put("metadata", d.metadata); + m.put("dlq_retry_count", d.dlqRetryCount); + return m; + } + + static Map metric(TaskMetric t) { + Map m = new LinkedHashMap<>(); + m.put("task_name", t.taskName); + m.put("job_id", t.jobId); + m.put("wall_time_ns", t.wallTimeNs); + m.put("memory_bytes", t.memoryBytes); + m.put("succeeded", t.succeeded); + m.put("recorded_at", t.recordedAt); + return m; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java new file mode 100644 index 00000000..1b339c82 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java @@ -0,0 +1,339 @@ +package org.byteveda.taskito.dashboard; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.Executors; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Collectors; +import org.byteveda.taskito.Job; +import org.byteveda.taskito.JobFilter; +import org.byteveda.taskito.JobStatus; +import org.byteveda.taskito.Queue; +import org.byteveda.taskito.TaskitoException; + +/** + * A read/action dashboard API + static-SPA server backed by a {@link Queue}, + * built on the JDK's {@code com.sun.net.httpserver}. The JSON contract is + * snake_case with Unix-millisecond timestamps (see {@link Contract}). + * + *

When a {@code token} is configured, {@code /api/*} (except + * {@code /api/auth/status}) requires it via {@code ?token=} or a cookie. + */ +public final class DashboardServer implements AutoCloseable { + private static final ObjectMapper JSON = new ObjectMapper(); + private static final String COOKIE = "taskito_token"; + private static final long DEFAULT_LIMIT = 50; + + private static final Pattern JOB = Pattern.compile("^/api/jobs/([^/]+)$"); + private static final Pattern CANCEL = Pattern.compile("^/api/jobs/([^/]+)/cancel$"); + private static final Pattern RETRY = Pattern.compile("^/api/dead-letters/([^/]+)/retry$"); + private static final Pattern PAUSE = Pattern.compile("^/api/queues/([^/]+)/pause$"); + private static final Pattern RESUME = Pattern.compile("^/api/queues/([^/]+)/resume$"); + + private final HttpServer server; + private final Queue queue; + private final String token; + private final Path staticDir; + + private DashboardServer(HttpServer server, Queue queue, String token, Path staticDir) { + this.server = server; + this.queue = queue; + this.token = token; + this.staticDir = staticDir; + } + + public static DashboardServer start(Queue queue, int port) throws IOException { + return start(queue, port, null, null); + } + + /** Start on {@code port} (0 = ephemeral). {@code token}/{@code staticDir} may be null. */ + public static DashboardServer start(Queue queue, int port, String token, String staticDir) + throws IOException { + HttpServer server = HttpServer.create(new InetSocketAddress(port), 0); + Path dir = staticDir == null ? null : Paths.get(staticDir).normalize(); + DashboardServer dashboard = new DashboardServer(server, queue, token, dir); + server.createContext("/", dashboard::dispatch); + server.setExecutor(Executors.newCachedThreadPool()); + server.start(); + return dashboard; + } + + public int port() { + return server.getAddress().getPort(); + } + + @Override + public void close() { + server.stop(0); + } + + private void dispatch(HttpExchange exchange) throws IOException { + try { + String path = exchange.getRequestURI().getPath(); + if (path.startsWith("/api/")) { + handleApi(exchange, path); + } else { + serveStatic(exchange, path); + } + } catch (RuntimeException e) { + respond(exchange, 500, error(e.getMessage() == null ? e.toString() : e.getMessage())); + } finally { + exchange.close(); + } + } + + private void handleApi(HttpExchange exchange, String path) throws IOException { + Map query = query(exchange); + if (path.equals("/api/auth/status")) { + respond(exchange, 200, Collections.singletonMap("auth_required", token != null)); + return; + } + if (!authorized(exchange, query)) { + respond(exchange, 401, error("unauthorized")); + return; + } + + String method = exchange.getRequestMethod(); + if ("GET".equals(method) && handleGet(exchange, path, query)) { + return; + } + if ("POST".equals(method) && handlePost(exchange, path)) { + return; + } + respond(exchange, 404, error("not found")); + } + + private boolean handleGet(HttpExchange exchange, String path, Map query) + throws IOException { + switch (path) { + case "/api/stats": + respond(exchange, 200, Contract.stats(queue.stats())); + return true; + case "/api/stats/queues": + respond(exchange, 200, statsByQueue()); + return true; + case "/api/queues/paused": + respond(exchange, 200, queue.listPausedQueues()); + return true; + case "/api/jobs": + respond(exchange, 200, listJobs(query)); + return true; + case "/api/dead-letters": + respond(exchange, 200, listDead(query)); + return true; + case "/api/metrics": + respond(exchange, 200, listMetrics(query)); + return true; + case "/api/workers": + respond(exchange, 200, listWorkers()); + return true; + default: + return getJob(exchange, path); + } + } + + private boolean getJob(HttpExchange exchange, String path) throws IOException { + Matcher m = JOB.matcher(path); + if (!m.matches()) { + return false; + } + Optional job = queue.getJob(m.group(1)); + if (job.isPresent()) { + respond(exchange, 200, Contract.job(job.get())); + } else { + respond(exchange, 404, error("no such job")); + } + return true; + } + + private boolean handlePost(HttpExchange exchange, String path) throws IOException { + Matcher cancel = CANCEL.matcher(path); + if (cancel.matches()) { + respond(exchange, 200, Collections.singletonMap("cancelled", queue.cancel(cancel.group(1)))); + return true; + } + Matcher retry = RETRY.matcher(path); + if (retry.matches()) { + respond(exchange, 200, Collections.singletonMap("id", queue.retryDead(retry.group(1)))); + return true; + } + Matcher pause = PAUSE.matcher(path); + if (pause.matches()) { + queue.pauseQueue(pause.group(1)); + respond(exchange, 200, Collections.singletonMap("ok", true)); + return true; + } + Matcher resume = RESUME.matcher(path); + if (resume.matches()) { + queue.resumeQueue(resume.group(1)); + respond(exchange, 200, Collections.singletonMap("ok", true)); + return true; + } + return false; + } + + private Map statsByQueue() { + Map out = new LinkedHashMap<>(); + queue.statsAllQueues().forEach((name, stats) -> out.put(name, Contract.stats(stats))); + return out; + } + + private List> listJobs(Map query) { + JobFilter.Builder filter = JobFilter.builder(); + if (query.containsKey("status")) { + filter.status(JobStatus.fromWire(query.get("status"))); + } + if (query.containsKey("queue")) { + filter.queue(query.get("queue")); + } + if (query.containsKey("task")) { + filter.task(query.get("task")); + } + if (query.containsKey("limit")) { + filter.limit(Integer.parseInt(query.get("limit"))); + } + if (query.containsKey("offset")) { + filter.offset(Integer.parseInt(query.get("offset"))); + } + return queue.listJobs(filter.build()).stream().map(Contract::job).collect(Collectors.toList()); + } + + private List> listDead(Map query) { + long limit = longParam(query, "limit", DEFAULT_LIMIT); + long offset = longParam(query, "offset", 0); + return queue.listDead(limit, offset).stream().map(Contract::dead).collect(Collectors.toList()); + } + + private List> listMetrics(Map query) { + long since = longParam(query, "since", 0); + return queue.metrics(query.get("task"), since).stream() + .map(Contract::metric) + .collect(Collectors.toList()); + } + + private List> listWorkers() { + return queue.listWorkers().stream().map(Contract::worker).collect(Collectors.toList()); + } + + private void serveStatic(HttpExchange exchange, String path) throws IOException { + if (staticDir == null) { + respond(exchange, 404, error("not found")); + return; + } + Path target = staticDir.resolve(path.substring(1)).normalize(); + if (!target.startsWith(staticDir) || !Files.isRegularFile(target)) { + target = staticDir.resolve("index.html"); + } + if (!Files.isRegularFile(target)) { + respond(exchange, 404, error("not found")); + return; + } + byte[] body = Files.readAllBytes(target); + exchange.getResponseHeaders().set("Content-Type", contentType(target)); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(body); + } + } + + private boolean authorized(HttpExchange exchange, Map query) { + if (token == null) { + return true; + } + if (token.equals(query.get("token"))) { + exchange.getResponseHeaders().add("Set-Cookie", COOKIE + "=" + token + "; HttpOnly; Path=/"); + return true; + } + return token.equals(cookieToken(exchange)); + } + + private static String cookieToken(HttpExchange exchange) { + List cookies = exchange.getRequestHeaders().getOrDefault("Cookie", Collections.emptyList()); + for (String header : cookies) { + for (String pair : header.split(";")) { + String trimmed = pair.trim(); + if (trimmed.startsWith(COOKIE + "=")) { + return trimmed.substring(COOKIE.length() + 1); + } + } + } + return null; + } + + private static long longParam(Map query, String key, long fallback) { + String value = query.get(key); + return value == null ? fallback : Long.parseLong(value); + } + + private static Map query(HttpExchange exchange) { + Map out = new HashMap<>(); + String raw = exchange.getRequestURI().getRawQuery(); + if (raw == null) { + return out; + } + for (String pair : raw.split("&")) { + int eq = pair.indexOf('='); + if (eq < 0) { + continue; + } + String key = URLDecoder.decode(pair.substring(0, eq), StandardCharsets.UTF_8); + String value = URLDecoder.decode(pair.substring(eq + 1), StandardCharsets.UTF_8); + out.put(key, value); + } + return out; + } + + private static Map error(String message) { + return Collections.singletonMap("error", message); + } + + private static void respond(HttpExchange exchange, int status, Object body) throws IOException { + byte[] out; + try { + out = JSON.writeValueAsBytes(body); + } catch (Exception e) { + throw new TaskitoException("failed to encode response", e); + } + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(status, out.length); + try (OutputStream stream = exchange.getResponseBody()) { + stream.write(out); + } + } + + private static String contentType(Path file) { + String name = file.getFileName().toString(); + if (name.endsWith(".html")) { + return "text/html; charset=utf-8"; + } + if (name.endsWith(".js")) { + return "application/javascript"; + } + if (name.endsWith(".css")) { + return "text/css"; + } + if (name.endsWith(".json")) { + return "application/json"; + } + if (name.endsWith(".svg")) { + return "image/svg+xml"; + } + return "application/octet-stream"; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/package-info.java new file mode 100644 index 00000000..d0db156c --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/package-info.java @@ -0,0 +1,2 @@ +/** Dashboard HTTP server (JDK httpserver) and the snake_case/Unix-ms JSON contract. */ +package org.byteveda.taskito.dashboard; From 4beb8d6e3181cb5fa48902d252bbaf6bace6a67e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:10:28 +0530 Subject: [PATCH 4/8] test(java): cover dashboard API and auth --- .../org/byteveda/taskito/DashboardTest.java | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/DashboardTest.java diff --git a/sdks/java/src/test/java/org/byteveda/taskito/DashboardTest.java b/sdks/java/src/test/java/org/byteveda/taskito/DashboardTest.java new file mode 100644 index 00000000..56632575 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/DashboardTest.java @@ -0,0 +1,65 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.file.Path; +import java.util.Collections; +import java.util.Map; +import org.byteveda.taskito.dashboard.DashboardServer; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +class DashboardTest { + + @SuppressWarnings("unchecked") + private static Class> mapType() { + return (Class>) (Class) Map.class; + } + + private static HttpResponse get(int port, String path) throws Exception { + return HttpClient.newHttpClient().send( + HttpRequest.newBuilder(URI.create("http://localhost:" + port + path)).GET().build(), + HttpResponse.BodyHandlers.ofString()); + } + + @Test + @Timeout(30) + void servesSnakeCaseContract(@TempDir Path dir) throws Exception { + Task> add = Task.of("add", mapType()); + try (Queue queue = + Taskito.builder().backend("sqlite").url(dir.resolve("t.db").toString()).open()) { + queue.enqueue(add, Collections.singletonMap("a", 1), + EnqueueOptions.builder().queue("emails").build()); + try (DashboardServer server = DashboardServer.start(queue, 0)) { + int port = server.port(); + + HttpResponse stats = get(port, "/api/stats"); + assertEquals(200, stats.statusCode()); + assertTrue(stats.body().contains("\"pending\":1")); + + HttpResponse jobs = get(port, "/api/jobs?queue=emails"); + assertTrue(jobs.body().contains("\"task_name\":\"add\"")); + assertTrue(jobs.body().contains("\"status\":\"pending\"")); + } + } + } + + @Test + @Timeout(30) + void enforcesToken(@TempDir Path dir) throws Exception { + try (Queue queue = + Taskito.builder().backend("sqlite").url(dir.resolve("t.db").toString()).open()) { + try (DashboardServer server = DashboardServer.start(queue, 0, "sekret", null)) { + int port = server.port(); + assertEquals(401, get(port, "/api/stats").statusCode()); + assertEquals(200, get(port, "/api/stats?token=sekret").statusCode()); + } + } + } +} From b9c9f3b7152442bd7599aefd532343022902ab95 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:10:28 +0530 Subject: [PATCH 5/8] docs(java): document middleware, dashboard, serializers --- sdks/java/README.md | 36 +++++++++++++++++++++++++++++++++--- 1 file changed, 33 insertions(+), 3 deletions(-) diff --git a/sdks/java/README.md b/sdks/java/README.md index abb77919..2bb8c725 100644 --- a/sdks/java/README.md +++ b/sdks/java/README.md @@ -3,9 +3,10 @@ A typed Java 11+ client over the Taskito Rust core, via a hand-written JNI shell (`crates/taskito-java`). -> Status: **build-out** (Phases 0–2). Producer + inspection + admin + logs, and -> worker task execution, are implemented and verified end-to-end against the -> native library. Dashboard, webhooks, and the CLI follow in later phases. +> Status: **build-out** (Phases 0–3 core). Producer + inspection + admin + logs, +> worker task execution, middleware, signed/encrypted serializers, and the +> dashboard server are implemented and verified end-to-end. Webhooks and the CLI +> follow next. ## Usage @@ -50,6 +51,35 @@ Runtime.getRuntime().addShutdownHook(new Thread(() -> { worker.awaitShutdown(); ``` +### Middleware + +```java +queue.use(new Middleware() { + @Override public void onEnqueue(EnqueueContext ctx) { /* validate / rewrite */ } + @Override public void before(TaskContext ctx) { /* trace */ } + @Override public void onDeadLetter(OutcomeEvent e) { /* alert */ } +}); +``` + +### Dashboard + +```java +try (DashboardServer dashboard = DashboardServer.start(queue, 8080, token, staticDir)) { + // GET /api/stats, /api/jobs, /api/workers, ... ; POST /api/jobs/{id}/cancel, ... + dashboard.port(); +} +``` + +### Serializers + +```java +byte[] key = ...; // 16/24/32 bytes for AES +Queue secure = Taskito.builder() + .backend("sqlite").url("taskito.db") + .serializer(new EncryptedSerializer(new JsonSerializer(), key)) + .open(); +``` + ## Structure ```text From c16f5211d7c28b97f2004383f0924671694da845 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:10:29 +0530 Subject: [PATCH 6/8] feat(java): add webhooks WebhookManager (CRUD + Middleware) persists subscriptions via the settings store and delivers matching job outcomes over HTTP with an HMAC-SHA256 signature and retries. --- .../byteveda/taskito/webhooks/Deliverer.java | 62 +++++++++ .../byteveda/taskito/webhooks/Webhook.java | 120 +++++++++++++++++ .../taskito/webhooks/WebhookManager.java | 124 ++++++++++++++++++ .../taskito/webhooks/WebhookStore.java | 41 ++++++ .../taskito/webhooks/package-info.java | 2 + 5 files changed, 349 insertions(+) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/webhooks/Webhook.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookStore.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/webhooks/package-info.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java new file mode 100644 index 00000000..9e40119e --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Deliverer.java @@ -0,0 +1,62 @@ +package org.byteveda.taskito.webhooks; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import org.byteveda.taskito.TaskitoException; + +/** Delivers webhook payloads over HTTP with an optional HMAC signature and retries. */ +final class Deliverer { + private static final String ALGORITHM = "HmacSHA256"; + private static final char[] HEX = "0123456789abcdef".toCharArray(); + + private final HttpClient client = HttpClient.newHttpClient(); + + /** Send {@code body} to the hook (non-blocking). Retries server errors. */ + void deliver(Webhook hook, byte[] body) { + HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(hook.url)) + .timeout(Duration.ofMillis(hook.timeoutMs)) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofByteArray(body)); + if (hook.secret != null) { + request.header("X-Taskito-Signature", "sha256=" + sign(hook.secret, body)); + } + hook.headers.forEach(request::header); + sendWithRetry(request.build(), hook.maxRetries); + } + + private void sendWithRetry(HttpRequest request, int attemptsLeft) { + client.sendAsync(request, HttpResponse.BodyHandlers.discarding()) + .whenComplete((response, error) -> { + boolean failed = error != null || response.statusCode() >= 500; + if (failed && attemptsLeft > 0) { + sendWithRetry(request, attemptsLeft - 1); + } + }); + } + + static String sign(String secret, byte[] body) { + try { + Mac mac = Mac.getInstance(ALGORITHM); + mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), ALGORITHM)); + return hex(mac.doFinal(body)); + } catch (Exception e) { + throw new TaskitoException("webhook signature failed", e); + } + } + + private static String hex(byte[] bytes) { + char[] out = new char[bytes.length * 2]; + for (int i = 0; i < bytes.length; i++) { + int b = bytes[i] & 0xff; + out[i * 2] = HEX[b >>> 4]; + out[i * 2 + 1] = HEX[b & 0x0f]; + } + return new String(out); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Webhook.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Webhook.java new file mode 100644 index 00000000..2e54d4fa --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/Webhook.java @@ -0,0 +1,120 @@ +package org.byteveda.taskito.webhooks; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.byteveda.taskito.events.EventName; + +/** A stored webhook subscription. Timestamps are Unix milliseconds. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class Webhook { + public final String id; + public final String url; + /** Outcome wire names this hook fires on: success/retry/dead/cancelled. */ + public final List events; + public final String taskFilter; + public final Map headers; + public final String secret; + public final int maxRetries; + public final long timeoutMs; + public final boolean enabled; + public final String description; + public final long createdAt; + public final long updatedAt; + + @JsonCreator + Webhook( + @JsonProperty("id") String id, + @JsonProperty("url") String url, + @JsonProperty("events") List events, + @JsonProperty("taskFilter") String taskFilter, + @JsonProperty("headers") Map headers, + @JsonProperty("secret") String secret, + @JsonProperty("maxRetries") int maxRetries, + @JsonProperty("timeoutMs") long timeoutMs, + @JsonProperty("enabled") boolean enabled, + @JsonProperty("description") String description, + @JsonProperty("createdAt") long createdAt, + @JsonProperty("updatedAt") long updatedAt) { + this.id = id; + this.url = url; + this.events = events == null ? Collections.emptyList() : events; + this.taskFilter = taskFilter; + this.headers = headers == null ? Collections.emptyMap() : headers; + this.secret = secret; + this.maxRetries = maxRetries; + this.timeoutMs = timeoutMs; + this.enabled = enabled; + this.description = description; + this.createdAt = createdAt; + this.updatedAt = updatedAt; + } + + public static Builder builder(String url) { + return new Builder(url); + } + + /** A draft webhook; the manager assigns its id and timestamps on create. */ + public static final class Builder { + final String url; + final List events = new ArrayList<>(); + final Map headers = new LinkedHashMap<>(); + String taskFilter; + String secret; + int maxRetries = 3; + long timeoutMs = 10_000; + boolean enabled = true; + String description; + + Builder(String url) { + this.url = url; + } + + public Builder on(EventName... names) { + for (EventName name : names) { + events.add(name.name().toLowerCase(java.util.Locale.ROOT)); + } + return this; + } + + public Builder taskFilter(String taskFilter) { + this.taskFilter = taskFilter; + return this; + } + + public Builder header(String name, String value) { + headers.put(name, value); + return this; + } + + public Builder secret(String secret) { + this.secret = secret; + return this; + } + + public Builder maxRetries(int maxRetries) { + this.maxRetries = maxRetries; + return this; + } + + public Builder timeoutMs(long timeoutMs) { + this.timeoutMs = timeoutMs; + return this; + } + + public Builder enabled(boolean enabled) { + this.enabled = enabled; + return this; + } + + public Builder description(String description) { + this.description = description; + return this; + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java new file mode 100644 index 00000000..4ad8a02e --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookManager.java @@ -0,0 +1,124 @@ +package org.byteveda.taskito.webhooks; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import org.byteveda.taskito.Queue; +import org.byteveda.taskito.TaskitoException; +import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.middleware.Middleware; + +/** + * Manages webhook subscriptions and dispatches matching job outcomes to them. + * {@link #attach} registers the manager as queue middleware so outcomes are + * delivered automatically. Persisted via the queue's settings store. + */ +public final class WebhookManager implements Middleware { + private static final ObjectMapper JSON = new ObjectMapper(); + + private final WebhookStore store; + private final Deliverer deliverer = new Deliverer(); + + private WebhookManager(Queue queue) { + this.store = new WebhookStore(queue); + } + + /** Create a manager and register it on {@code queue} for automatic dispatch. */ + public static WebhookManager attach(Queue queue) { + WebhookManager manager = new WebhookManager(queue); + queue.use(manager); + return manager; + } + + public synchronized Webhook create(Webhook.Builder spec) { + long now = System.currentTimeMillis(); + Webhook hook = new Webhook( + UUID.randomUUID().toString(), + spec.url, + new ArrayList<>(spec.events), + spec.taskFilter, + new LinkedHashMap<>(spec.headers), + spec.secret, + spec.maxRetries, + spec.timeoutMs, + spec.enabled, + spec.description, + now, + now); + List all = store.load(); + all.add(hook); + store.save(all); + return hook; + } + + public List list() { + return store.load(); + } + + public Optional get(String id) { + return store.load().stream().filter(hook -> hook.id.equals(id)).findFirst(); + } + + public synchronized boolean delete(String id) { + List all = store.load(); + boolean removed = all.removeIf(hook -> hook.id.equals(id)); + if (removed) { + store.save(all); + } + return removed; + } + + @Override + public void onCompleted(OutcomeEvent event) { + dispatch(event); + } + + @Override + public void onRetry(OutcomeEvent event) { + dispatch(event); + } + + @Override + public void onDeadLetter(OutcomeEvent event) { + dispatch(event); + } + + @Override + public void onCancel(OutcomeEvent event) { + dispatch(event); + } + + private void dispatch(OutcomeEvent event) { + String wire = event.name.name().toLowerCase(Locale.ROOT); + byte[] body = payload(event, wire); + for (Webhook hook : store.load()) { + if (hook.enabled && hook.events.contains(wire) && matches(hook.taskFilter, event.taskName)) { + deliverer.deliver(hook, body); + } + } + } + + private static boolean matches(String filter, String taskName) { + return filter == null || filter.equals(taskName); + } + + private static byte[] payload(OutcomeEvent event, String wire) { + Map body = new LinkedHashMap<>(); + body.put("event", wire); + body.put("job_id", event.jobId); + body.put("task_name", event.taskName); + body.put("error", event.error); + body.put("retry_count", event.retryCount); + body.put("timed_out", event.timedOut); + try { + return JSON.writeValueAsBytes(body); + } catch (Exception e) { + throw new TaskitoException("webhook payload encoding failed", e); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookStore.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookStore.java new file mode 100644 index 00000000..c5cdb9e5 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/WebhookStore.java @@ -0,0 +1,41 @@ +package org.byteveda.taskito.webhooks; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.type.CollectionType; +import java.util.ArrayList; +import java.util.List; +import org.byteveda.taskito.Queue; +import org.byteveda.taskito.TaskitoException; + +/** Persists webhooks as a JSON list in the queue's settings key/value store. */ +final class WebhookStore { + private static final String KEY = "taskito.webhooks"; + private static final ObjectMapper JSON = new ObjectMapper(); + + private final Queue queue; + + WebhookStore(Queue queue) { + this.queue = queue; + } + + List load() { + return queue.getSetting(KEY).map(WebhookStore::parse).orElseGet(ArrayList::new); + } + + void save(List webhooks) { + try { + queue.setSetting(KEY, JSON.writeValueAsString(webhooks)); + } catch (Exception e) { + throw new TaskitoException("failed to persist webhooks", e); + } + } + + private static List parse(String json) { + try { + CollectionType type = JSON.getTypeFactory().constructCollectionType(List.class, Webhook.class); + return JSON.readValue(json, type); + } catch (Exception e) { + throw new TaskitoException("failed to read webhooks", e); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/webhooks/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/package-info.java new file mode 100644 index 00000000..ac9c2eb0 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/webhooks/package-info.java @@ -0,0 +1,2 @@ +/** Webhook subscriptions: WebhookManager (CRUD + outcome dispatch), persisted via settings, delivered with HMAC. */ +package org.byteveda.taskito.webhooks; From dbaf3ae3a6e012b23199343acbf5ef01ce3be3fc Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:10:29 +0530 Subject: [PATCH 7/8] feat(java): add picocli CLI Commands: stats, enqueue, jobs, cancel, pause/resume, dlq, dashboard. --- sdks/java/build.gradle.kts | 1 + .../java/org/byteveda/taskito/cli/Cli.java | 263 ++++++++++++++++++ .../byteveda/taskito/cli/package-info.java | 2 + 3 files changed, 266 insertions(+) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/cli/package-info.java diff --git a/sdks/java/build.gradle.kts b/sdks/java/build.gradle.kts index a2a17239..1de930bf 100644 --- a/sdks/java/build.gradle.kts +++ b/sdks/java/build.gradle.kts @@ -43,6 +43,7 @@ tasks.withType().configureEach { dependencies { api("com.fasterxml.jackson.core:jackson-databind:2.17.2") + implementation("info.picocli:picocli:4.7.6") testImplementation(platform("org.junit:junit-bom:5.10.3")) testImplementation("org.junit.jupiter:junit-jupiter") diff --git a/sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java b/sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java new file mode 100644 index 00000000..48fe8b52 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java @@ -0,0 +1,263 @@ +package org.byteveda.taskito.cli; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import org.byteveda.taskito.DeadJob; +import org.byteveda.taskito.Job; +import org.byteveda.taskito.JobFilter; +import org.byteveda.taskito.JobStatus; +import org.byteveda.taskito.Queue; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.dashboard.DashboardServer; +import picocli.CommandLine; +import picocli.CommandLine.Command; +import picocli.CommandLine.Option; +import picocli.CommandLine.Parameters; +import picocli.CommandLine.ParentCommand; + +/** Command-line interface over a Taskito queue. */ +@Command( + name = "taskito", + mixinStandardHelpOptions = true, + subcommands = { + Cli.Stats.class, + Cli.Enqueue.class, + Cli.Jobs.class, + Cli.Cancel.class, + Cli.Pause.class, + Cli.Resume.class, + Cli.Dlq.class, + Cli.Dashboard.class + }) +public final class Cli { + static final ObjectMapper JSON = new ObjectMapper(); + + @Option(names = "--backend", description = "Storage backend (default sqlite).", defaultValue = "sqlite") + String backend; + + @Option(names = "--url", required = true, description = "Connection string (SQLite path or URL).") + String url; + + Queue open() { + return Taskito.builder().backend(backend).url(url).open(); + } + + static String json(Object value) { + try { + return JSON.writerWithDefaultPrettyPrinter().writeValueAsString(value); + } catch (Exception e) { + return String.valueOf(value); + } + } + + public static void main(String[] args) { + System.exit(new CommandLine(new Cli()).execute(args)); + } + + @Command(name = "stats", description = "Show job counts by status.") + static final class Stats implements Callable { + @ParentCommand + Cli parent; + + @Override + public Integer call() { + try (Queue queue = parent.open()) { + System.out.println(json(queue.stats())); + } + return 0; + } + } + + @Command(name = "enqueue", description = "Enqueue a task with a JSON payload.") + static final class Enqueue implements Callable { + @ParentCommand + Cli parent; + + @Parameters(index = "0", description = "Task name.") + String task; + + @Parameters(index = "1", arity = "0..1", description = "JSON payload (default null).") + String payload; + + @Override + public Integer call() throws Exception { + Object value = payload == null ? null : JSON.readValue(payload, Object.class); + try (Queue queue = parent.open()) { + System.out.println(queue.enqueue(task, value)); + } + return 0; + } + } + + @Command(name = "jobs", description = "List jobs.") + static final class Jobs implements Callable { + @ParentCommand + Cli parent; + + @Option(names = "--status", description = "Filter by status.") + String status; + + @Option(names = "--queue", description = "Filter by queue.") + String queue; + + @Option(names = "--limit", defaultValue = "50") + int limit; + + @Override + public Integer call() { + JobFilter.Builder filter = JobFilter.builder().limit(limit); + if (status != null) { + filter.status(JobStatus.fromWire(status)); + } + if (queue != null) { + filter.queue(queue); + } + try (Queue q = parent.open()) { + List jobs = q.listJobs(filter.build()); + System.out.println(json(jobs)); + } + return 0; + } + } + + @Command(name = "cancel", description = "Cancel a pending job.") + static final class Cancel implements Callable { + @ParentCommand + Cli parent; + + @Parameters(index = "0", description = "Job id.") + String id; + + @Override + public Integer call() { + try (Queue queue = parent.open()) { + return queue.cancel(id) ? 0 : 1; + } + } + } + + @Command(name = "pause", description = "Pause a queue.") + static final class Pause implements Callable { + @ParentCommand + Cli parent; + + @Parameters(index = "0", description = "Queue name.") + String queue; + + @Override + public Integer call() { + try (Queue q = parent.open()) { + q.pauseQueue(queue); + } + return 0; + } + } + + @Command(name = "resume", description = "Resume a queue.") + static final class Resume implements Callable { + @ParentCommand + Cli parent; + + @Parameters(index = "0", description = "Queue name.") + String queue; + + @Override + public Integer call() { + try (Queue q = parent.open()) { + q.resumeQueue(queue); + } + return 0; + } + } + + @Command( + name = "dlq", + description = "Dead-letter operations.", + subcommands = {Dlq.ListDead.class, Dlq.Retry.class, Dlq.Delete.class}) + static final class Dlq { + @ParentCommand + Cli parent; + + Queue open() { + return parent.open(); + } + + @Command(name = "list", description = "List dead-letter entries.") + static final class ListDead implements Callable { + @ParentCommand + Dlq dlq; + + @Option(names = "--limit", defaultValue = "50") + long limit; + + @Override + public Integer call() { + try (Queue queue = dlq.open()) { + List dead = queue.listDead(limit, 0); + System.out.println(json(dead)); + } + return 0; + } + } + + @Command(name = "retry", description = "Re-enqueue a dead-letter entry.") + static final class Retry implements Callable { + @ParentCommand + Dlq dlq; + + @Parameters(index = "0", description = "Dead-letter id.") + String id; + + @Override + public Integer call() { + try (Queue queue = dlq.open()) { + System.out.println(queue.retryDead(id)); + } + return 0; + } + } + + @Command(name = "delete", description = "Delete a dead-letter entry.") + static final class Delete implements Callable { + @ParentCommand + Dlq dlq; + + @Parameters(index = "0", description = "Dead-letter id.") + String id; + + @Override + public Integer call() { + try (Queue queue = dlq.open()) { + return queue.deleteDead(id) ? 0 : 1; + } + } + } + } + + @Command(name = "dashboard", description = "Serve the dashboard until interrupted.") + static final class Dashboard implements Callable { + @ParentCommand + Cli parent; + + @Option(names = "--port", defaultValue = "8080") + int port; + + @Option(names = "--token", description = "Require this token for API access.") + String token; + + @Option(names = "--static", description = "Directory of the prebuilt SPA.") + String staticDir; + + @Override + public Integer call() throws Exception { + try (Queue queue = parent.open(); + DashboardServer server = DashboardServer.start(queue, port, token, staticDir)) { + System.out.println("dashboard on http://localhost:" + server.port()); + new CountDownLatch(1).await(); + } + return 0; + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/cli/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/cli/package-info.java new file mode 100644 index 00000000..de4b059f --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/cli/package-info.java @@ -0,0 +1,2 @@ +/** Command-line interface (picocli): stats, enqueue, jobs, cancel, pause/resume, dlq, dashboard. */ +package org.byteveda.taskito.cli; From 4dda0a8f839d6ec1c1d7e6989aa7e6054c74220b Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:10:29 +0530 Subject: [PATCH 8/8] test(java): cover webhook CRUD --- .../org/byteveda/taskito/WebhookTest.java | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/WebhookTest.java diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WebhookTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WebhookTest.java new file mode 100644 index 00000000..95eb4f72 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WebhookTest.java @@ -0,0 +1,38 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.webhooks.Webhook; +import org.byteveda.taskito.webhooks.WebhookManager; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class WebhookTest { + + @Test + void crudRoundTrip(@TempDir Path dir) { + try (Queue queue = + Taskito.builder().backend("sqlite").url(dir.resolve("t.db").toString()).open()) { + WebhookManager webhooks = WebhookManager.attach(queue); + + Webhook created = webhooks.create(Webhook.builder("http://example/hook") + .on(EventName.SUCCESS, EventName.DEAD) + .secret("s")); + assertNotNull(created.id); + assertEquals(2, created.events.size()); + + assertEquals(1, webhooks.list().size()); + assertTrue(webhooks.get(created.id).isPresent()); + assertTrue(webhooks.get(created.id).get().events.contains("success")); + + assertTrue(webhooks.delete(created.id)); + assertTrue(webhooks.list().isEmpty()); + assertFalse(webhooks.delete(created.id)); + } + } +}