fix: count limit, persist snapshot, JSON dequeue

README.md

@@ -52,7 +52,7 @@ To get the next payload from the `foo` queue, send a get request to `/dequeue/:q
 curl -X GET http://127.0.0.1:1991/dequeue/foo
 ```
 
-This returns the oldest added payload on queue `foo` and removes it, guaranteeing both the order and that each payload will only be read once.
+This returns the oldest added payload on queue `foo` as JSON and removes it, guaranteeing both the order and that each payload will only be read once. Strings, numbers, booleans, arrays, and objects all use `application/json` so a string `"0"` is distinct from the number `0`.
 
 That's all you need to get started! 😎
 
@@ -113,7 +113,7 @@ To get persistency, simply add the `--persist` option when starting up the serve
 docker run -d -e PORT=1991 -e HOST=0.0.0.0 -e PERSIST=/mnt/ jonbaldie/queue /usr/bin/queue --persist
 ```
 
-If the server sees that the `persist.dat` file exists on startup, it will run the binary log from the beginning and then clear the file down.
+If the server sees that the `persist.dat` file exists on startup, it will replay the binary log and then rewrite the file as a snapshot of remaining items.
 
 When using Docker, it might be useful to add `persist.dat` as a persistent volume to keep your binary logs safe.
 

src/handler.ts

@@ -67,12 +67,9 @@ function itemResponse(item: unknown): Response {
     if (item === undefined) {
         return new Response(null, { status: 204 });
     }
-    if (typeof item === "object" && item !== null) {
-        return new Response(JSON.stringify(item), {
-            headers: { "Content-Type": "application/json" },
-        });
-    }
-    return new Response(String(item));
+    return new Response(JSON.stringify(item), {
+        headers: { "Content-Type": "application/json" },
+    });
 }
 
 function dequeueHandler(mgr: QueueManager<string>): RouteHandler {

src/manager.ts

@@ -83,10 +83,6 @@ export default class Manager<T = string> {
         }
     }
 
-    private registered(name: string): boolean {
-        return this.queues.has(name);
-    }
-
     public canCreateQueue(): boolean {
         return this.queues.size < this.queueCountLimit;
     }
@@ -95,10 +91,8 @@ export default class Manager<T = string> {
         this.validateName(name);
         const queue = this.find(name);
         if (!queue) {
-            // Creating a new queue - check if we have room
-            return this.canCreateQueue();
+            return this.canCreateQueue() && 0 < this.queueDepthLimit;
         }
-        // Existing queue - check if it has room
         return queue.length < this.queueDepthLimit;
     }
 
@@ -108,15 +102,17 @@ export default class Manager<T = string> {
 
     public enqueue(name: string, payload: T): Manager<T> {
         this.validateName(name);
-        const queue = this.find(name) || new FIFOQueue<T>();
-
-        if (this.registered(name) === false) {
-            this.register(name, queue);
+        const existing = this.find(name);
+        if (!existing && !this.canCreateQueue()) {
+            throw new Error("Queue count limit reached");
         }
-
+        const queue = existing || new FIFOQueue<T>();
         if (queue.length >= this.queueDepthLimit) {
             throw new Error("Queue depth limit reached");
         }
+        if (!existing) {
+            this.register(name, queue);
+        }
         queue.push(payload);
 
         if (this.persistEnabled) {
@@ -183,7 +179,7 @@ export default class Manager<T = string> {
         for (const event of this.store.loadState()) {
             this.applyLoadedEvent(event);
         }
-        this.store.clear();
+        this.save();
     }
 
     private applyLoadedEvent(event: QueueEvent<T>): void {

tests/handler_test.ts

@@ -747,8 +747,8 @@ Deno.test("GET peek returns 200 with payload when queue has items", async () =>
     });
     const response = await handler(request);
     assertEquals(response.status, 200);
-    const body = await response.text();
-    assertEquals(body, "peekable");
+    assertEquals(response.headers.get("content-type"), "application/json");
+    assertEquals(await response.json(), "peekable");
 });
 
 // Happy path: GET peek returns 204 when queue is empty
@@ -806,7 +806,7 @@ Deno.test("dequeue returns application/json for array payload", async () => {
     assertEquals(body, [1, 2, 3]);
 });
 
-Deno.test("dequeue returns text/plain for number payload", async () => {
+Deno.test("dequeue returns application/json for number payload", async () => {
     const mgr = new QueueManager(new Persistency.MemoryStore);
     const handler = createHandler(mgr, API_TOKEN);
 
@@ -823,11 +823,11 @@ Deno.test("dequeue returns text/plain for number payload", async () => {
     });
     const response = await handler(request);
     assertEquals(response.status, 200);
-    const body = await response.text();
-    assertEquals(body, "42");
+    assertEquals(response.headers.get("content-type"), "application/json");
+    assertEquals(await response.json(), 42);
 });
 
-Deno.test("dequeue returns text/plain for boolean payload", async () => {
+Deno.test("dequeue returns application/json for boolean payload", async () => {
     const mgr = new QueueManager(new Persistency.MemoryStore);
     const handler = createHandler(mgr, API_TOKEN);
 
@@ -844,8 +844,8 @@ Deno.test("dequeue returns text/plain for boolean payload", async () => {
     });
     const response = await handler(request);
     assertEquals(response.status, 200);
-    const body = await response.text();
-    assertEquals(body, "true");
+    assertEquals(response.headers.get("content-type"), "application/json");
+    assertEquals(await response.json(), true);
 });
 
 // queue-nyc: Missing payload key returns 400
@@ -1003,7 +1003,7 @@ Deno.test("body read error returns 413 not 400", async () => {
     assertEquals(response.status, 413);
 });
 
-Deno.test("dequeue returns text/plain for string payload", async () => {
+Deno.test("dequeue returns application/json for string payload", async () => {
     const mgr = new QueueManager(new Persistency.MemoryStore);
     const handler = createHandler(mgr, API_TOKEN);
 
@@ -1020,8 +1020,8 @@ Deno.test("dequeue returns text/plain for string payload", async () => {
     });
     const response = await handler(request);
     assertEquals(response.status, 200);
-    const body = await response.text();
-    assertEquals(body, "hello world");
+    assertEquals(response.headers.get("content-type"), "application/json");
+    assertEquals(await response.json(), "hello world");
 });
 
 Deno.test("Manager: enqueue followed by multiple dequeues (catches state corruption)", () => {
@@ -1145,8 +1145,7 @@ Deno.test("API: enqueue stores payload and dequeue retrieves it (catches data lo
         })
     );
     assertEquals(deqRes.status, 200);
-    const retrieved = await deqRes.text();
-    assertEquals(retrieved, payload);
+    assertEquals(await deqRes.json(), payload);
 });
 
 Deno.test("API: dequeue empty queue returns 204 (catches crash)", async () => {
@@ -1271,8 +1270,7 @@ Deno.test("API: FIFO order through HTTP (catches dequeue order mutation)", async
                 headers: { "Authorization": `Bearer ${TEST_TOKEN}` },
             })
         );
-        const actual = await res.text();
-        assertEquals(actual, expected);
+        assertEquals(await res.json(), expected);
     }
 });
 
@@ -1297,13 +1295,13 @@ Deno.test("API: multiple queues isolated (catches queue mixing)", async () => {
             headers: { "Authorization": `Bearer ${TEST_TOKEN}` },
         })
     );
-    assertEquals(await res1.text(), "q1-item");
+    assertEquals(await res1.json(), "q1-item");
     const res2 = await handler(
         new Request("http://localhost/dequeue/q2", {
             headers: { "Authorization": `Bearer ${TEST_TOKEN}` },
         })
     );
-    assertEquals(await res2.text(), "q2-item");
+    assertEquals(await res2.json(), "q2-item");
 });
 
 Deno.test("API: enqueue with empty payload (catches validation)", async () => {
@@ -1347,8 +1345,7 @@ Deno.test("API: very long payload (catches buffer handling)", async () => {
             headers: { "Authorization": `Bearer ${TEST_TOKEN}` },
         })
     );
-    const retrieved = await deqRes.text();
-    assertEquals(retrieved, longPayload);
+    assertEquals(await deqRes.json(), longPayload);
 });
 
 Deno.test("auth: no token on enqueue returns 401", async () => {
@@ -1449,3 +1446,51 @@ Deno.test("auth: valid token on length returns 200", async () => {
     const res = await handler(req);
     assertEquals(200, res.status);
 });
+
+Deno.test("dequeue distinguishes string zero from number zero", async () => {
+    const mgr = new QueueManager(new Persistency.MemoryStore);
+    const handler = createHandler(mgr, API_TOKEN);
+
+    await handler(new Request("http://localhost:3000/enqueue/q", {
+        method: "POST",
+        body: JSON.stringify({ payload: "0" }),
+        headers: authHeaders,
+    }));
+    await handler(new Request("http://localhost:3000/enqueue/q", {
+        method: "POST",
+        body: JSON.stringify({ payload: 0 }),
+        headers: authHeaders,
+    }));
+
+    const first = await handler(new Request("http://localhost:3000/dequeue/q", {
+        headers: authHeaders,
+    }));
+    assertEquals(first.status, 200);
+    assertEquals(first.headers.get("content-type"), "application/json");
+    assertEquals(await first.json(), "0");
+
+    const second = await handler(new Request("http://localhost:3000/dequeue/q", {
+        headers: authHeaders,
+    }));
+    assertEquals(second.status, 200);
+    assertEquals(second.headers.get("content-type"), "application/json");
+    assertEquals(await second.json(), 0);
+});
+
+Deno.test("peek returns JSON for a string payload", async () => {
+    const mgr = new QueueManager(new Persistency.MemoryStore);
+    const handler = createHandler(mgr, API_TOKEN);
+
+    await handler(new Request("http://localhost:3000/enqueue/q", {
+        method: "POST",
+        body: JSON.stringify({ payload: "hello" }),
+        headers: authHeaders,
+    }));
+
+    const response = await handler(new Request("http://localhost:3000/peek/q", {
+        headers: authHeaders,
+    }));
+    assertEquals(response.status, 200);
+    assertEquals(response.headers.get("content-type"), "application/json");
+    assertEquals(await response.json(), "hello");
+});

tests/manager_test.ts

@@ -194,7 +194,9 @@ Deno.test("manager persistency", () => {
     const mgr = new QueueManager(persist);
     mgr.load();
 
-    assertEquals([], persist.loadState());
+    const snapshot = persist.loadState();
+    assertEquals(snapshot.length, 2);
+    assertEquals(snapshot.every((event) => event.enqueue === true), true);
     assertEquals(1, mgr.length("foo"));
     assertEquals("bar", mgr.dequeue("foo"));
     assertEquals(1, mgr.length("fee"));
@@ -212,7 +214,10 @@ Deno.test("json persistency", () => {
 
     mgr.load();
 
-    assertEquals([], persist.loadState());
+    const snapshot = persist.loadState();
+    assertEquals(snapshot.length, 1);
+    assertEquals(snapshot[0].payload, payload);
+    assertEquals(snapshot[0].enqueue, true);
     assertEquals(1, mgr.length("foo"));
     assertEquals(payload, mgr.dequeue("foo"));
     persist.close();
@@ -577,3 +582,85 @@ Deno.test("manager load skips events with neither enqueue nor dequeue", () => {
     assertEquals(mgr.length("q"), 1);
     assertEquals(mgr.dequeue("q"), "a");
 });
+
+Deno.test("enqueue throws when queue count limit reached", () => {
+    const mgr = new QueueManager(new Persistency.MemoryStore(), 10, 1);
+    mgr.enqueue("a", "1");
+    assertThrows(() => mgr.enqueue("b", "2"), Error, "Queue count limit reached");
+    assertEquals(mgr.listQueues(), ["a"]);
+    assertEquals(mgr.dequeue("a"), "1");
+});
+
+Deno.test("enqueue on an existing queue is allowed at the count limit", () => {
+    const mgr = new QueueManager(new Persistency.MemoryStore(), 10, 1);
+    mgr.enqueue("a", "1");
+    mgr.enqueue("a", "2");
+    assertEquals(mgr.length("a"), 2);
+    assertEquals(mgr.dequeue("a"), "1");
+    assertEquals(mgr.dequeue("a"), "2");
+});
+
+Deno.test("canEnqueue false means enqueue does not create a queue", () => {
+    const mgr = new QueueManager(new Persistency.MemoryStore(), 10, 1);
+    mgr.enqueue("a", "1");
+    assertEquals(mgr.canEnqueue("b"), false);
+    assertThrows(() => mgr.enqueue("b", "2"), Error, "Queue count limit reached");
+    assertEquals(mgr.listQueues(), ["a"]);
+});
+
+Deno.test("depth limit 0 refuses enqueue without leaking an empty queue", () => {
+    const mgr = new QueueManager(new Persistency.MemoryStore(), 0, 2);
+    assertEquals(mgr.canEnqueue("q"), false);
+    assertThrows(() => mgr.enqueue("q", "x"), Error, "Queue depth limit reached");
+    assertEquals(mgr.listQueues(), []);
+    assertEquals(mgr.length("q"), 0);
+});
+
+Deno.test("load snapshots remaining items so a second load restores them", () => {
+    const persist = new Persistency.MemoryStore();
+    const mgr = new QueueManager(persist);
+    mgr.enqueue("jobs", "keep-me");
+
+    const boot = new Persistency.MemoryStore();
+    for (const ev of persist.loadState()) {
+        boot.saveEvent(ev.queue, ev.payload, ev.enqueue);
+    }
+    const loaded = new QueueManager(boot);
+    loaded.load();
+    assertEquals(loaded.peek("jobs"), "keep-me");
+
+    const boot2 = new Persistency.MemoryStore();
+    for (const ev of boot.loadState()) {
+        boot2.saveEvent(ev.queue, ev.payload, ev.enqueue);
+    }
+    const loaded2 = new QueueManager(boot2);
+    loaded2.load();
+    assertEquals(loaded2.dequeue("jobs"), "keep-me");
+});
+
+Deno.test("FileStore load snapshots remaining items onto disk", () => {
+    const tmp = Deno.makeTempDirSync();
+    try {
+        const store = new Persistency.FileStore();
+        store.dir(tmp);
+        const mgr = new QueueManager(store);
+        mgr.enqueue("jobs", "keep-me");
+        store.close();
+
+        const store2 = new Persistency.FileStore();
+        store2.dir(tmp);
+        const loaded = new QueueManager(store2);
+        loaded.load();
+        assertEquals(loaded.peek("jobs"), "keep-me");
+        store2.close();
+
+        const store3 = new Persistency.FileStore();
+        store3.dir(tmp);
+        const loaded2 = new QueueManager(store3);
+        loaded2.load();
+        assertEquals(loaded2.dequeue("jobs"), "keep-me");
+        store3.close();
+    } finally {
+        Deno.removeSync(tmp, { recursive: true });
+    }
+});