Jobs and webhooks

JOB=$(curl -s http://127.0.0.1:7341/v1/jobs \
  -H "Authorization: Bearer $SUPERPDF_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "pdf_run_batch",
    "input": {
      "inputs": ["report.pdf", "appendix.pdf"],
      "steps": [{ "action": "watermark", "options": { "text": "DRAFT" } }],
      "output_folder": "drafts"
    }
  }' | jq -r .id)

# Check twice a second until the job has finished
while true; do
  STATUS=$(curl -s http://127.0.0.1:7341/v1/jobs/$JOB \
    -H "Authorization: Bearer $SUPERPDF_API_TOKEN" | tee job.json | jq -r .status)
  [ "$STATUS" = queued ] || [ "$STATUS" = running ] || break
  sleep 0.5
done
jq '.result // .error' job.json

import { request, type BatchResult, type Job } from './superpdf.ts';

let job = await request<Job<BatchResult>>('POST', '/v1/jobs', {
  tool: 'pdf_run_batch',
  input: {
    inputs: ['report.pdf', 'appendix.pdf'],
    steps: [{ action: 'watermark', options: { text: 'DRAFT' } }],
    output_folder: 'drafts',
  },
});

// Check twice a second until the job has finished
while (job.status === 'queued' || job.status === 'running') {
  await new Promise(resolve => setTimeout(resolve, 500));
  job = await request<Job<BatchResult>>('GET', `/v1/jobs/${job.id}`);
}
if (!job.result) throw new Error(job.error?.detail);
for (const file of job.result.outputs) console.log(`${file.input} -> ${file.output}`);

import time

from superpdf import BatchResult, Job, request

job: Job = request("POST", "/v1/jobs", {
    "tool": "pdf_run_batch",
    "input": {
        "inputs": ["report.pdf", "appendix.pdf"],
        "steps": [{"action": "watermark", "options": {"text": "DRAFT"}}],
        "output_folder": "drafts",
    },
})

# Check twice a second until the job has finished
while job["status"] in ("queued", "running"):
    time.sleep(0.5)
    job = request("GET", f"/v1/jobs/{job['id']}")
if job["status"] == "failed":
    raise SystemExit(job["error"]["detail"])
result: BatchResult = job["result"]
for file in result["outputs"]:
    print(f"{file['input']} -> {file['output']}")

package main

import (
	"fmt"
	"log"
	"time"
)

func main() {
	var job Job[BatchResult]
	err := Request("POST", "/v1/jobs", map[string]any{
		"tool": "pdf_run_batch",
		"input": map[string]any{
			"inputs": []string{"report.pdf", "appendix.pdf"},
			"steps": []map[string]any{
				{"action": "watermark", "options": map[string]string{"text": "DRAFT"}},
			},
			"output_folder": "drafts",
		},
	}, &job)
	if err != nil {
		log.Fatal(err)
	}

	// Check twice a second until the job has finished
	for job.Status == "queued" || job.Status == "running" {
		time.Sleep(500 * time.Millisecond)
		if err := Request("GET", "/v1/jobs/"+job.ID, nil, &job); err != nil {
			log.Fatal(err)
		}
	}
	if job.Result == nil {
		log.Fatal(job.Error)
	}
	for _, file := range job.Result.Outputs {
		fmt.Printf("%s -> %s\n", file.Input, file.Output)
	}
}

curl http://127.0.0.1:7341/v1/jobs \
  -H "Authorization: Bearer $SUPERPDF_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "pdf_run_batch",
    "input": {
      "inputs": ["report.pdf", "appendix.pdf"],
      "steps": [{ "action": "watermark", "options": { "text": "DRAFT" } }],
      "output_folder": "drafts"
    },
    "callback_url": "http://127.0.0.1:8080/hooks/superpdf"
  }'

import { request, type BatchResult, type Job } from './superpdf.ts';

const job = await request<Job<BatchResult>>('POST', '/v1/jobs', {
  tool: 'pdf_run_batch',
  input: {
    inputs: ['report.pdf', 'appendix.pdf'],
    steps: [{ action: 'watermark', options: { text: 'DRAFT' } }],
    output_folder: 'drafts',
  },
  callback_url: 'http://127.0.0.1:8080/hooks/superpdf',
});
console.log(`${job.id} is ${job.status}`);

from superpdf import Job, request

job: Job = request("POST", "/v1/jobs", {
    "tool": "pdf_run_batch",
    "input": {
        "inputs": ["report.pdf", "appendix.pdf"],
        "steps": [{"action": "watermark", "options": {"text": "DRAFT"}}],
        "output_folder": "drafts",
    },
    "callback_url": "http://127.0.0.1:8080/hooks/superpdf",
})
print(f"{job['id']} is {job['status']}")

package main

import (
	"fmt"
	"log"
)

func main() {
	var job Job[BatchResult]
	err := Request("POST", "/v1/jobs", map[string]any{
		"tool": "pdf_run_batch",
		"input": map[string]any{
			"inputs": []string{"report.pdf", "appendix.pdf"},
			"steps": []map[string]any{
				{"action": "watermark", "options": map[string]string{"text": "DRAFT"}},
			},
			"output_folder": "drafts",
		},
		"callback_url": "http://127.0.0.1:8080/hooks/superpdf",
	}, &job)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("%s is %s\n", job.ID, job.Status)
}

import { createServer, type IncomingHttpHeaders } from 'node:http';
import { createHmac, timingSafeEqual } from 'node:crypto';

interface JobEvent {
  type: 'job.succeeded' | 'job.failed';
  timestamp: string;
  data: {
    job_id: string;
    tool: string;
    status: 'succeeded' | 'failed';
    outputs: string[];
    error?: { code: string; detail: string };
  };
}

const secret = process.env.SUPERPDF_WEBHOOK_SECRET ?? '';
const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64');
const handled = new Set<string>();

/** Standard Webhooks: "v1," + base64 HMAC-SHA256 of "id.timestamp.body". */
function verify(headers: IncomingHttpHeaders, body: Buffer): boolean {
  const id = String(headers['webhook-id'] ?? '');
  const timestamp = String(headers['webhook-timestamp'] ?? '');
  if (!id || !/^\d+$/.test(timestamp)) return false;
  // Refuse events more than five minutes old (or ahead)
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false;
  const mac = createHmac('sha256', key).update(`${id}.${timestamp}.`).update(body);
  const expected = Buffer.from(`v1,${mac.digest('base64')}`);
  const signatures = String(headers['webhook-signature'] ?? '').split(' ');
  return signatures.some(signature => {
    const given = Buffer.from(signature);
    return given.length === expected.length && timingSafeEqual(given, expected);
  });
}

const server = createServer(async (req, res) => {
  if (req.method !== 'POST' || req.url !== '/hooks/superpdf') {
    res.writeHead(404).end();
    return;
  }
  const chunks: Buffer[] = [];
  for await (const chunk of req) chunks.push(chunk);
  const body = Buffer.concat(chunks);
  if (!verify(req.headers, body)) {
    res.writeHead(401).end();
    return;
  }
  res.writeHead(204).end(); // answer at once: superPDF waits 10 seconds at most

  const id = String(req.headers['webhook-id']);
  if (handled.has(id)) return; // a retry of an event already handled
  handled.add(id);
  const event: JobEvent = JSON.parse(body.toString('utf8'));
  const files = event.data.outputs.length;
  console.log(`${event.type} ${event.data.job_id}: ${files} file(s)`);
});
server.listen(8080, '127.0.0.1', () => {
  console.log('Listening on http://127.0.0.1:8080/hooks/superpdf');
});

import base64
import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

secret = os.environ["SUPERPDF_WEBHOOK_SECRET"]
key = base64.b64decode(secret.removeprefix("whsec_"))
handled: set[str] = set()


def verify(headers, body: bytes) -> bool:
    """Standard Webhooks: "v1," + base64 HMAC-SHA256 of "id.timestamp.body"."""
    msg_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "")
    if not msg_id or not timestamp.isascii() or not timestamp.isdigit():
        return False
    # Refuse events more than five minutes old (or ahead)
    if abs(time.time() - int(timestamp)) > 300:
        return False
    signed = f"{msg_id}.{timestamp}.".encode() + body
    digest = hmac.new(key, signed, hashlib.sha256).digest()
    expected = b"v1," + base64.b64encode(digest)
    signatures = headers.get("webhook-signature", "").split(" ")
    return any(hmac.compare_digest(s.encode(), expected) for s in signatures)


class Receiver(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        if self.path != "/hooks/superpdf":
            self.send_error(404)
            return
        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        if not verify(self.headers, body):
            self.send_response(401)
            self.end_headers()
            return
        self.send_response(204)  # answer at once: superPDF waits 10 seconds at most
        self.end_headers()

        msg_id = self.headers["webhook-id"]
        if msg_id in handled:  # a retry of an event already handled
            return
        handled.add(msg_id)
        event = json.loads(body)
        files = len(event["data"]["outputs"])
        print(f"{event['type']} {event['data']['job_id']}: {files} file(s)")


server = HTTPServer(("127.0.0.1", 8080), Receiver)
print("Listening on http://127.0.0.1:8080/hooks/superpdf")
server.serve_forever()

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

var (
	key     []byte
	mu      sync.Mutex
	handled = map[string]bool{}
)

// verify checks a Standard Webhooks signature:
// "v1," + base64 HMAC-SHA256 of "id.timestamp.body".
func verify(h http.Header, body []byte) bool {
	id, timestamp := h.Get("webhook-id"), h.Get("webhook-timestamp")
	sent, err := strconv.ParseInt(timestamp, 10, 64)
	if id == "" || err != nil {
		return false
	}
	// Refuse events more than five minutes old (or ahead)
	if age := time.Now().Unix() - sent; age > 300 || age < -300 {
		return false
	}
	mac := hmac.New(sha256.New, key)
	fmt.Fprintf(mac, "%s.%s.", id, timestamp)
	mac.Write(body)
	expected := "v1," + base64.StdEncoding.EncodeToString(mac.Sum(nil))
	for _, signature := range strings.Split(h.Get("webhook-signature"), " ") {
		if hmac.Equal([]byte(signature), []byte(expected)) {
			return true
		}
	}
	return false
}

func receive(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
	if err != nil || !verify(r.Header, body) {
		w.WriteHeader(http.StatusUnauthorized)
		return
	}
	w.WriteHeader(http.StatusNoContent)

	id := r.Header.Get("webhook-id")
	mu.Lock()
	seen := handled[id]
	handled[id] = true
	mu.Unlock()
	if seen {
		return // a retry of an event already handled
	}
	var event struct {
		Type string `json:"type"`
		Data struct {
			JobID   string   `json:"job_id"`
			Outputs []string `json:"outputs"`
		} `json:"data"`
	}
	if err := json.Unmarshal(body, &event); err != nil {
		log.Print(err)
		return
	}
	files := len(event.Data.Outputs)
	fmt.Printf("%s %s: %d file(s)\n", event.Type, event.Data.JobID, files)
}

func main() {
	secret := os.Getenv("SUPERPDF_WEBHOOK_SECRET")
	var err error
	key, err = base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
	if err != nil {
		log.Fatal("SUPERPDF_WEBHOOK_SECRET is not a whsec_ secret")
	}
	listener, err := net.Listen("tcp", "127.0.0.1:8080")
	if err != nil {
		log.Fatal(err)
	}
	http.HandleFunc("POST /hooks/superpdf", receive)
	fmt.Println("Listening on http://127.0.0.1:8080/hooks/superpdf")
	log.Fatal(http.Serve(listener, nil))
}

node receiver.ts

python3 receiver.py

go run receiver.go