Generic ack worker

A minimal worker that implements the SimpleQ ack-mode protocol with no downstream provider SDK — just SimpleQ's own @simpleq/sdk in Node, plain HTTP everywhere else. It receives a webhook, verifies the signature, responds 200 immediately, runs the work out of band, and reports back with ack / nack / defer.

This is the shortest path to a working ack-mode worker. Drop in your own downstream call and you have the integration.

Reach for ack mode over standard mode when the work can outlive the 15-second window a synchronous webhook response must fit in: the 200 returned up front only tells SimpleQ the delivery landed, and the job stays open until your worker reports what actually happened. The routing here maps downstream HTTP status onto that report — a 429, 503, or 529 becomes a defer that relays the downstream Retry-After so SimpleQ holds and redelivers the job without burning an attempt, while other 4xx map to a non-retryable nack and 5xx or network errors to a retryable one. If the process crashes between the 200 and the report, the queue's ackTimeout with ackTimeoutAction: "retry" redelivers the job, so nothing is silently dropped.

Prerequisite

This example requires a queue created with mode: 'ack'. Standard-mode queues use a different flow — see Ack mode for how the two differ.

The worker

Pick the tab for your language — each tab is a complete, single-file worker you can copy and run.

// npm install @simpleq/sdk express
import express from 'express';
import { SimpleQ, retryAfterSeconds } from '@simpleq/sdk';
import { simpleqWebhookHandler } from '@simpleq/sdk/express';

const simpleq = new SimpleQ({ apiKey: process.env.SIMPLEQ_API_KEY });

const app = express();

// Replace this with your real work.
// Throw an error with a numeric `.status` property to route HTTP-shaped failures:
//   429 / 503 / 529 → defer (backpressure, no attempt burned)
//   4xx            → nack non-retryable (hard failure)
//   5xx / network  → nack retryable (retried with backoff)
async function doTheWork(payload) {
  // your code here
}

// simpleqWebhookHandler verifies x-simpleq-signature against the raw body (401 on
// mismatch). Ack mode: send the 200 yourself, then do the work out of band and
// report the outcome via ack / nack / defer.
app.post(
  '/webhook',
  simpleqWebhookHandler(process.env.SQ_SIGNING_SECRET, async (job, { res }) => {
    res.status(200).end(); // ack mode: respond now, do the work after

    try {
      await doTheWork(job.payload);
      await simpleq.ack(job.id);
    } catch (err) {
      const status = err?.status;

      if (status === 429 || status === 503 || status === 529) {
        // Backpressure: held and redelivered, no attempt burned. Relay Retry-After
        // when present; 10s fallback covers responses without it (e.g. 529).
        await simpleq.defer(job.id, { retryAfter: retryAfterSeconds(err) ?? 10, reason: `downstream ${status}` });
      } else if (status >= 400 && status < 500) {
        await simpleq.nack(job.id, { retryable: false, reason: `downstream ${status}` });
      } else {
        await simpleq.nack(job.id, {
          retryable: true,
          reason: `downstream ${status ?? 'network'}: ${err?.message ?? 'unknown'}`,
        });
      }
    }
  }),
);

app.listen(9000, () => {
  console.log('generic ack worker on :9000');
});

python

import hashlib
import hmac
import json
import os

import httpx
from fastapi import BackgroundTasks, FastAPI, Header, Request, Response

SQ_API_URL = os.environ["SQ_API_URL"]
SIMPLEQ_API_KEY = os.environ["SIMPLEQ_API_KEY"]
SQ_SIGNING_SECRET = os.environ["SQ_SIGNING_SECRET"]

app = FastAPI()


# Raise this from do_the_work to route HTTP-shaped downstream failures.
class DownstreamError(Exception):
    def __init__(self, status=None, message="", retry_after=None):
        super().__init__(message or f"downstream {status}")
        self.status = status
        self.retry_after = retry_after


# Verify the webhook came from SimpleQ.
# See https://docs.simpleq.io/concepts/signature-verification for the full breakdown.
def verify_signature(raw_body: bytes, header: str | None) -> bool:
    if not header:
        return False
    expected = "sha256=" + hmac.new(SQ_SIGNING_SECRET.encode(), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header)


# Send /ack, /nack, or /defer back to SimpleQ.
# Reuse one client across callbacks for connection pooling.
_client = httpx.AsyncClient(timeout=10.0)


async def callback(job_id: str, kind: str, body: dict | None = None) -> None:
    res = await _client.post(
        f"{SQ_API_URL}/v1/jobs/{job_id}/{kind}",
        headers={
            "content-type": "application/json",
            "authorization": f"Bearer {SIMPLEQ_API_KEY}",
        },
        json=body or {},
    )
    if res.is_error:
        print(f"[{job_id}] {kind} failed: {res.status_code} {res.text}")


# Replace this with your real work.
# Raise DownstreamError with a numeric `status` to route HTTP-shaped failures:
#   429 / 503 / 529 → defer (backpressure, no attempt burned)
#   4xx            → nack non-retryable (hard failure)
#   5xx / network  → nack retryable (retried with backoff)
async def do_the_work(payload: dict) -> None:
    # your code here
    pass


@app.post("/webhook")
async def webhook(
    request: Request,
    background_tasks: BackgroundTasks,
    x_simpleq_signature: str | None = Header(default=None),
):
    # Read the raw body BEFORE parsing — the HMAC is over the exact bytes.
    raw = await request.body()
    if not verify_signature(raw, x_simpleq_signature):
        return Response(status_code=401)

    # Ack mode: 200 immediately, then run the work out of band.
    job = json.loads(raw)
    # BackgroundTasks runs in-process — if this worker crashes mid-job the
    # work is lost, but ackTimeout + ackTimeoutAction: "retry" redelivers it.
    background_tasks.add_task(process_job, job)
    return Response(status_code=200)


async def process_job(job: dict) -> None:
    try:
        await do_the_work(job["payload"])
        await callback(job["id"], "ack")
    except DownstreamError as err:
        status = err.status

        if status in (429, 503, 529):
            # 529 (overloaded) carries no Retry-After → falls back to 10s.
            await callback(job["id"], "defer", {
                "retryAfter": err.retry_after or 10,
                "reason": f"downstream {status}",
            })
        elif status is not None and 400 <= status < 500:
            await callback(job["id"], "nack", {"retryable": False, "reason": f"downstream {status}"})
        else:
            await callback(job["id"], "nack", {
                "retryable": True,
                "reason": f"downstream {status or 'network'}: {err}",
            })
    except Exception as err:
        # Any non-HTTP-shaped error → retry with backoff.
        await callback(job["id"], "nack", {"retryable": True, "reason": f"downstream network: {err}"})


# Run with: uvicorn worker:app --port 9000

package main

import (
	"bytes"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
)

var (
	sqAPIURL        = os.Getenv("SQ_API_URL")
	sqAPIKey        = os.Getenv("SIMPLEQ_API_KEY")
	sqSigningSecret = os.Getenv("SQ_SIGNING_SECRET")
)

// A SimpleQ job as delivered to your webhook. Your data is under Payload.
type Job struct {
	ID          string          `json:"id"`
	Queue       string          `json:"queue"`
	Payload     json.RawMessage `json:"payload"`
	Attempt     int             `json:"attempt"`
	MaxAttempts int             `json:"maxAttempts"`
	CreatedAt   string          `json:"createdAt"`
}

// Return this from doTheWork to route HTTP-shaped downstream failures.
type downstreamError struct {
	status     int // HTTP status from the downstream call (0 = network)
	message    string
	retryAfter int // seconds; 0 = none
}

func (e *downstreamError) Error() string {
	if e.message != "" {
		return e.message
	}
	return fmt.Sprintf("downstream %d", e.status)
}

// Verify the webhook came from SimpleQ.
// See https://docs.simpleq.io/concepts/signature-verification for the full breakdown.
func verifySignature(rawBody []byte, header string) bool {
	if header == "" {
		return false
	}
	mac := hmac.New(sha256.New, []byte(sqSigningSecret))
	mac.Write(rawBody)
	expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))
	return hmac.Equal([]byte(expected), []byte(header))
}

// Send /ack, /nack, or /defer back to SimpleQ.
func callback(jobID, kind string, body map[string]any) {
	if body == nil {
		body = map[string]any{}
	}
	buf, _ := json.Marshal(body)
	req, _ := http.NewRequest("POST", fmt.Sprintf("%s/v1/jobs/%s/%s", sqAPIURL, jobID, kind), bytes.NewReader(buf))
	req.Header.Set("content-type", "application/json")
	req.Header.Set("authorization", "Bearer "+sqAPIKey)

	res, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Printf("[%s] %s failed: %v", jobID, kind, err)
		return
	}
	defer res.Body.Close()
	if res.StatusCode >= 400 {
		msg, _ := io.ReadAll(res.Body)
		log.Printf("[%s] %s failed: %d %s", jobID, kind, res.StatusCode, msg)
	}
}

// Replace this with your real work.
// Return a *downstreamError with a numeric status to route HTTP-shaped failures:
//   429 / 503 / 529 → defer (backpressure, no attempt burned)
//   4xx            → nack non-retryable (hard failure)
//   5xx / network  → nack retryable (retried with backoff)
func doTheWork(payload json.RawMessage) error {
	// your code here
	return nil
}

func webhook(w http.ResponseWriter, r *http.Request) {
	// Read the raw body BEFORE parsing — the HMAC is over the exact bytes.
	raw, err := io.ReadAll(r.Body)
	if err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	if !verifySignature(raw, r.Header.Get("x-simpleq-signature")) {
		w.WriteHeader(http.StatusUnauthorized)
		return
	}

	var job Job
	if err := json.Unmarshal(raw, &job); err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	// Ack mode: 200 immediately, then run the work out of band.
	w.WriteHeader(http.StatusOK)
	go processJob(job)
}

func processJob(job Job) {
	err := doTheWork(job.Payload)
	if err == nil {
		callback(job.ID, "ack", nil)
		return
	}

	de, ok := err.(*downstreamError)
	if !ok {
		// Any non-HTTP-shaped error → retry with backoff.
		callback(job.ID, "nack", map[string]any{"retryable": true, "reason": "downstream network: " + err.Error()})
		return
	}

	status := de.status
	switch {
	case status == 429 || status == 503 || status == 529:
		retryAfter := de.retryAfter // 529 carries no Retry-After → falls back to 10s
		if retryAfter == 0 {
			retryAfter = 10
		}
		callback(job.ID, "defer", map[string]any{
			"retryAfter": retryAfter,
			"reason":     fmt.Sprintf("downstream %d", status),
		})
	case status >= 400 && status < 500:
		callback(job.ID, "nack", map[string]any{"retryable": false, "reason": fmt.Sprintf("downstream %d", status)})
	default:
		label := strconv.Itoa(status)
		if status == 0 {
			label = "network"
		}
		callback(job.ID, "nack", map[string]any{
			"retryable": true,
			"reason":    fmt.Sprintf("downstream %s: %s", label, de.Error()),
		})
	}
}

func main() {
	http.HandleFunc("/webhook", webhook)
	log.Printf("generic ack worker on :9000 → %s", sqAPIURL)
	log.Fatal(http.ListenAndServe(":9000", nil))
}

java

// Spring Boot single-file ack worker. Deps (Maven):
//   org.springframework.boot:spring-boot-starter-web
//   com.fasterxml.jackson.core:jackson-databind  (bundled with starter-web)
// Java 17+ (uses java.net.http.HttpClient).
package com.example.simpleq;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RestController;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
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.security.MessageDigest;
import java.util.LinkedHashMap;
import java.util.Map;

@SpringBootApplication
@EnableAsync
public class Worker {
    public static void main(String[] args) {
        // This worker example listens on port 9000 (set server.port=9000).
        SpringApplication.run(Worker.class, args);
        System.out.println("generic ack worker on :9000 -> " + System.getenv("SQ_API_URL"));
    }
}

@RestController
class WebhookController {
    static final String SQ_SIGNING_SECRET = System.getenv("SQ_SIGNING_SECRET");
    static final ObjectMapper MAPPER = new ObjectMapper();

    private final JobProcessor processor;

    WebhookController(JobProcessor processor) {
        this.processor = processor;
    }

    // Verify the webhook came from SimpleQ.
    // See https://docs.simpleq.io/concepts/signature-verification for the full breakdown.
    boolean verifySignature(byte[] rawBody, String header) {
        if (header == null) return false;
        String expected = "sha256=" + hmacHex(rawBody);
        // Constant-time compare over the raw bytes.
        return MessageDigest.isEqual(
            expected.getBytes(StandardCharsets.UTF_8),
            header.getBytes(StandardCharsets.UTF_8));
    }

    static String hmacHex(byte[] rawBody) {
        try {
            Mac mac = Mac.getInstance("HmacSHA256");
            mac.init(new SecretKeySpec(SQ_SIGNING_SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
            byte[] digest = mac.doFinal(rawBody);
            StringBuilder sb = new StringBuilder(digest.length * 2);
            for (byte b : digest) sb.append(String.format("%02x", b));
            return sb.toString();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @PostMapping(value = "/webhook", consumes = MediaType.ALL_VALUE)
    ResponseEntity<Void> webhook(
            @RequestBody(required = false) byte[] rawBody,
            @RequestHeader(value = "x-simpleq-signature", required = false) String signature) {
        // Read the raw body BEFORE parsing -- the HMAC is over the exact bytes.
        byte[] raw = rawBody == null ? new byte[0] : rawBody;
        if (!verifySignature(raw, signature)) {
            return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
        }

        // Ack mode: 200 immediately, then run the work out of band.
        JsonNode job;
        try {
            job = MAPPER.readTree(raw);
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.BAD_REQUEST).build();
        }
        // Call through the injected proxy bean so @Async actually takes effect
        // (a same-class self-invocation would bypass the proxy and run synchronously).
        processor.processJob(job);
        return ResponseEntity.ok().build();
    }
}

@Component
class JobProcessor {
    static final String SQ_API_URL = System.getenv("SQ_API_URL");
    static final String SIMPLEQ_API_KEY = System.getenv("SIMPLEQ_API_KEY");

    static final ObjectMapper MAPPER = new ObjectMapper();
    static final HttpClient HTTP = HttpClient.newHttpClient();

    // Raise this from doTheWork to route HTTP-shaped downstream failures.
    static class DownstreamError extends RuntimeException {
        final Integer status;
        final Integer retryAfter;
        DownstreamError(Integer status, String message, Integer retryAfter) {
            super(message != null && !message.isEmpty() ? message : "downstream " + status);
            this.status = status;
            this.retryAfter = retryAfter;
        }
    }

    // Send /ack, /nack, or /defer back to SimpleQ.
    void callback(String jobId, String kind, Map<String, Object> body) {
        try {
            String json = MAPPER.writeValueAsString(body == null ? Map.of() : body);
            HttpRequest req = HttpRequest.newBuilder()
                .uri(URI.create(SQ_API_URL + "/v1/jobs/" + jobId + "/" + kind))
                .header("content-type", "application/json")
                .header("authorization", "Bearer " + SIMPLEQ_API_KEY)
                .POST(HttpRequest.BodyPublishers.ofString(json))
                .build();
            HttpResponse<String> res = HTTP.send(req, HttpResponse.BodyHandlers.ofString());
            if (res.statusCode() >= 400) {
                System.err.printf("[%s] %s failed: %d %s%n", jobId, kind, res.statusCode(), res.body());
            }
        } catch (Exception err) {
            System.err.printf("[%s] %s failed: %s%n", jobId, kind, err.getMessage());
        }
    }

    // Replace this with your real work.
    // Throw DownstreamError with a numeric `status` to route HTTP-shaped failures:
    //   429 / 503 / 529 -> defer (backpressure, no attempt burned)
    //   4xx            -> nack non-retryable (hard failure)
    //   5xx / network  -> nack retryable (retried with backoff)
    void doTheWork(JsonNode payload) {
        // your code here
    }

    // @Async runs the work on a separate thread so the 200 returns first.
    @Async
    void processJob(JsonNode job) {
        String jobId = job.path("id").asText();
        try {
            doTheWork(job.path("payload"));
            callback(jobId, "ack", Map.of());
        } catch (DownstreamError err) {
            Integer status = err.status;

            if (status != null && (status == 429 || status == 503 || status == 529)) {
                // 529 (overloaded) carries no Retry-After → falls back to 10s.
                Map<String, Object> body = new LinkedHashMap<>();
                body.put("retryAfter", err.retryAfter != null ? err.retryAfter : 10);
                body.put("reason", "downstream " + status);
                callback(jobId, "defer", body);
            } else if (status != null && status >= 400 && status < 500) {
                Map<String, Object> body = new LinkedHashMap<>();
                body.put("retryable", false);
                body.put("reason", "downstream " + status);
                callback(jobId, "nack", body);
            } else {
                Map<String, Object> body = new LinkedHashMap<>();
                body.put("retryable", true);
                body.put("reason", "downstream " + (status != null ? status : "network") + ": " + err.getMessage());
                callback(jobId, "nack", body);
            }
        } catch (Exception err) {
            // Any non-HTTP-shaped error -> retry with backoff.
            Map<String, Object> body = new LinkedHashMap<>();
            body.put("retryable", true);
            body.put("reason", "downstream network: " + err.getMessage());
            callback(jobId, "nack", body);
        }
    }
}

php

<?php
// Generic ack worker — plain PHP webhook handler (no framework).
// Map your web server so that POST /webhook routes to this file.

$SQ_API_URL = getenv('SQ_API_URL');
$SIMPLEQ_API_KEY = getenv('SIMPLEQ_API_KEY');
$SQ_SIGNING_SECRET = getenv('SQ_SIGNING_SECRET');

// Raise this from do_the_work to route HTTP-shaped downstream failures.
class DownstreamError extends Exception {
    public $status;
    public $retry_after;
    public function __construct($status = null, $message = '', $retry_after = null) {
        parent::__construct($message !== '' ? $message : "downstream $status");
        $this->status = $status;
        $this->retry_after = $retry_after;
    }
}

// Verify the webhook came from SimpleQ.
// See https://docs.simpleq.io/concepts/signature-verification for the full breakdown.
function verify_signature($raw_body, $header) {
    global $SQ_SIGNING_SECRET;
    if (!$header) return false;
    $expected = 'sha256=' . hash_hmac('sha256', $raw_body, $SQ_SIGNING_SECRET);
    // hash_equals is constant-time — never use == on a signature.
    return hash_equals($expected, $header);
}

// Send /ack, /nack, or /defer back to SimpleQ.
function callback($job_id, $kind, $body = []) {
    global $SQ_API_URL, $SIMPLEQ_API_KEY;
    $ch = curl_init("$SQ_API_URL/v1/jobs/$job_id/$kind");
    curl_setopt($ch, CURLOPT_POST, true);
    curl_setopt($ch, CURLOPT_HTTPHEADER, [
        'content-type: application/json',
        "authorization: Bearer $SIMPLEQ_API_KEY",
    ]);
    // empty body must still be an object: {} not []
    curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode((object) $body));
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
    $res = curl_exec($ch);
    $code = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
    curl_close($ch);
    if ($code < 200 || $code >= 300) {
        error_log("[$job_id] $kind failed: $code $res");
    }
}

// Replace this with your real work.
// Throw DownstreamError with a numeric `status` to route HTTP-shaped failures:
//   429 / 503 / 529 → defer (backpressure, no attempt burned)
//   4xx            → nack non-retryable (hard failure)
//   5xx / network  → nack retryable (retried with backoff)
function do_the_work($payload) {
    // your code here
}

// Read the raw body BEFORE parsing — the HMAC is over the exact bytes.
$raw = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_SIMPLEQ_SIGNATURE'] ?? null;

if (!verify_signature($raw, $header)) {
    http_response_code(401);
    exit;
}

// Ack mode: 200 immediately, then run the work out of band.
http_response_code(200);
// KEY PHP GOTCHA: under PHP-FPM, fastcgi_finish_request() flushes the 200
// response to SimpleQ and closes the connection NOW, while this script keeps
// running. Without it the response would block on do_the_work below and could
// blow the 15s ack ceiling. (On non-FPM SAPIs this function is unavailable —
// run worker.php under PHP-FPM for true out-of-band processing.)
if (function_exists('fastcgi_finish_request')) {
    fastcgi_finish_request();
}

$job = json_decode($raw, true);

try {
    do_the_work($job['payload']);
    callback($job['id'], 'ack');
} catch (DownstreamError $err) {
    $status = $err->status;

    if ($status === 429 || $status === 503 || $status === 529) {
        // 529 (overloaded) carries no Retry-After → falls back to 10s.
        callback($job['id'], 'defer', [
            'retryAfter' => $err->retry_after ?: 10,
            'reason' => "downstream $status",
        ]);
    } elseif ($status !== null && $status >= 400 && $status < 500) {
        callback($job['id'], 'nack', ['retryable' => false, 'reason' => "downstream $status"]);
    } else {
        callback($job['id'], 'nack', [
            'retryable' => true,
            'reason' => 'downstream ' . ($status ?? 'network') . ': ' . $err->getMessage(),
        ]);
    }
} catch (Throwable $err) {
    // Any non-HTTP-shaped error → retry with backoff.
    callback($job['id'], 'nack', ['retryable' => true, 'reason' => 'downstream network: ' . $err->getMessage()]);
}

csharp

using System.Security.Cryptography;
using System.Text;
using System.Text.Json;

var SQ_API_URL = Environment.GetEnvironmentVariable("SQ_API_URL")!;
var SIMPLEQ_API_KEY = Environment.GetEnvironmentVariable("SIMPLEQ_API_KEY")!;
var SQ_SIGNING_SECRET = Environment.GetEnvironmentVariable("SQ_SIGNING_SECRET")!;

var http = new HttpClient();

var builder = WebApplication.CreateBuilder(args);
var app = builder.Build();

// Verify the webhook came from SimpleQ.
// See https://docs.simpleq.io/concepts/signature-verification for the full breakdown.
bool VerifySignature(byte[] rawBody, string? header)
{
    if (string.IsNullOrEmpty(header)) return false;
    using var hmac = new HMACSHA256(Encoding.UTF8.GetBytes(SQ_SIGNING_SECRET));
    var digest = Convert.ToHexString(hmac.ComputeHash(rawBody)).ToLowerInvariant();
    var expected = Encoding.UTF8.GetBytes("sha256=" + digest);
    var actual = Encoding.UTF8.GetBytes(header);
    // Constant-time compare; FixedTimeEquals also handles the length check.
    return CryptographicOperations.FixedTimeEquals(expected, actual);
}

// Send /ack, /nack, or /defer back to SimpleQ.
async Task Callback(string jobId, string kind, object? body = null)
{
    var req = new HttpRequestMessage(HttpMethod.Post, $"{SQ_API_URL}/v1/jobs/{jobId}/{kind}")
    {
        Content = new StringContent(JsonSerializer.Serialize(body ?? new { }), Encoding.UTF8, "application/json"),
    };
    req.Headers.Add("Authorization", $"Bearer {SIMPLEQ_API_KEY}");
    var res = await http.SendAsync(req);
    if (!res.IsSuccessStatusCode)
        Console.Error.WriteLine($"[{jobId}] {kind} failed: {(int)res.StatusCode} {await res.Content.ReadAsStringAsync()}");
}

// Replace this with your real work.
// Throw DownstreamException with a numeric `Status` to route HTTP-shaped failures:
//   429 / 503 / 529 → defer (backpressure, no attempt burned)
//   4xx            → nack non-retryable (hard failure)
//   5xx / network  → nack retryable (retried with backoff)
async Task DoTheWork(JsonElement payload)
{
    // your code here
    await Task.CompletedTask;
}

async Task ProcessJob(JsonElement job)
{
    var jobId = job.GetProperty("id").GetString()!;
    try
    {
        await DoTheWork(job.GetProperty("payload"));
        await Callback(jobId, "ack");
    }
    catch (DownstreamException err)
    {
        var status = err.Status;
        // 529 (overloaded) carries no Retry-After → falls back to 10s.
        if (status == 429 || status == 503 || status == 529)
            await Callback(jobId, "defer", new { retryAfter = err.RetryAfter ?? 10, reason = $"downstream {status}" });
        else if (status is >= 400 and < 500)
            await Callback(jobId, "nack", new { retryable = false, reason = $"downstream {status}" });
        else
            await Callback(jobId, "nack", new { retryable = true, reason = $"downstream {(status?.ToString() ?? "network")}: {err.Message}" });
    }
    catch (Exception err)
    {
        // Any non-HTTP-shaped error → retry with backoff.
        await Callback(jobId, "nack", new { retryable = true, reason = $"downstream network: {err.Message}" });
    }
}

app.MapPost("/webhook", async (HttpRequest request) =>
{
    // Read the raw body BEFORE parsing — the HMAC is over the exact bytes.
    using var ms = new MemoryStream();
    await request.Body.CopyToAsync(ms);
    var raw = ms.ToArray();

    var header = request.Headers["x-simpleq-signature"].FirstOrDefault();
    if (!VerifySignature(raw, header))
        return Results.Unauthorized();

    // Ack mode: 200 immediately, then run the work out of band.
    var job = JsonSerializer.Deserialize<JsonElement>(raw);
    // Fire-and-forget: survives the response but NOT a process crash —
    // ackTimeout + ackTimeoutAction: "retry" redelivers anything lost.
    _ = Task.Run(async () =>
    {
        try { await ProcessJob(job); }
        catch (Exception err) { Console.Error.WriteLine($"[{job.GetProperty("id").GetString()}] unhandled: {err}"); }
    });
    return Results.Ok();
});

Console.WriteLine($"generic ack worker on :9000 → {SQ_API_URL}");
app.Run("http://0.0.0.0:9000");

// Raise this from DoTheWork to route HTTP-shaped downstream failures.
// NOTE: type declarations must come AFTER all top-level statements.
class DownstreamException : Exception
{
    public int? Status { get; }
    public int? RetryAfter { get; }
    public DownstreamException(int? status = null, string message = "", int? retryAfter = null)
        : base(string.IsNullOrEmpty(message) ? $"downstream {status}" : message)
    {
        Status = status;
        RetryAfter = retryAfter;
    }
}

Run it

bash

npm init -y
npm i express
SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
node worker.mjs

bash

pip install fastapi uvicorn httpx
SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
uvicorn worker:app --port 9000

bash

SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
go run worker.go

bash

# spring-boot-starter-web on the classpath; server.port=9000
SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
./mvnw spring-boot:run -Dspring-boot.run.arguments=--server.port=9000

bash

# True out-of-band processing needs PHP-FPM behind nginx/Caddy (so
# fastcgi_finish_request() exists). For a quick local test:
SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
php -S 0.0.0.0:9000 worker.php

bash

SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
dotnet run

SQ_API_URL — the SimpleQ API base your worker calls back to: https://api.simpleq.io.
SIMPLEQ_API_KEY — create one on the dashboard's API Keys page.
SQ_SIGNING_SECRET — your queue's signing secret. See Signature verification for where to find it (dashboard reveal or POST /v1/queues response, depending on how the queue was created).

Your worker listens on port 9000, but SimpleQ delivers from the cloud and can't reach localhost. Open a tunnel and point the queue at it:

bash

cloudflared tunnel --url http://localhost:9000

Set the printed https://… URL (with /webhook appended) as your queue's webhookUrl. See Local development for the full flow and updating the URL on restart.

What each piece does

Piece	Concept
Verifying `x-simpleq-signature` over the raw body	Signature verification — prevents forged webhooks.
Responding `200` before processing	Ack mode — acknowledges receipt under the 15s ceiling.
`ack` callback on success	The job is marked complete.
`defer` callback on 429 / 503 / 529	Backpressure — hold and redeliver, no attempt burned.
`nack` `{ retryable: false }` callback on 4xx	Hard failure. Routed to the DLQ if the queue has `dlqEnabled: true`; otherwise dead-ended.
`nack` `{ retryable: true }` callback on 5xx / network	Transient failure → retry with backoff.

Test it with a publish

Once the worker is running, publish a job to trigger a delivery:

bash

curl -X POST "$SQ_API_URL/v1/queues/<your-queue>/jobs" \
  -H "Authorization: Bearer $SIMPLEQ_API_KEY" \
  -H "content-type: application/json" \
  -d '{ "payload": { "hello": "world" }, "idempotencyKey": "test-1" }'

You should see a 200 on the worker side and the job transition to completed (assuming doTheWork succeeded). Watch the queue detail page in the dashboard to see the state change in real time.

Generic ack worker ​

The worker ​

Run it ​

What each piece does ​

Test it with a publish ​

Generic ack worker

The worker

Run it

What each piece does

Test it with a publish