feat(java): serializers, middleware, dashboard, webhooks, CLI

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

sdks/java/build.gradle.kts

@@ -43,6 +43,7 @@ tasks.withType<Checkstyle>().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")

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> 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 <T> String enqueue(Task<T> task, T payload) {
 
     @Override
     public <T> String enqueue(Task<T> 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<TaskLog> getTaskLogs(String jobId) {
 
     @Override
     public Worker.Builder worker() {
-        return Worker.builder(backend, serializer);
+        return Worker.builder(backend, serializer, middleware);
     }
 
     @Override

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. */

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> 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> middleware;
         private final Map<String, RegisteredTask> handlers = new HashMap<>();
         private final Map<EventName, List<Consumer<OutcomeEvent>>> 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> middleware) {
             this.backend = backend;
             this.serializer = serializer;
+            this.middleware = middleware;
         }
 
         public <T, R> Builder handle(String taskName, Class<T> payloadType, TaskFunction<T, R> 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);

sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java

@@ -1,37 +1,44 @@
 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<String, RegisteredTask> handlers;
     private final Serializer serializer;
     private final ExecutorService executor;
     private final Emitter emitter;
+    private final List<Middleware> middleware;
     // Resolved once startWorker returns; job tasks await it before completing.
     private final CompletableFuture<WorkerControl> control = new CompletableFuture<>();
 
     WorkerDispatchBridge(
             Map<String, RegisteredTask> handlers,
             Serializer serializer,
             ExecutorService executor,
-            Emitter emitter) {
+            Emitter emitter,
+            List<Middleware> 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) {