Tasks API

cURL

# cURL is for trying the API by hand; the other tabs are complete clients.
KEY="$MAILMAN_TASKS_KEY"
HOST="https://tasks.getmailman.app"

# Check the key. -i prints the status line and headers (Retry-After, for example).
curl --connect-timeout 5 --max-time 15 -i "$HOST/api/v1/user/verify" -H "Authorization: Token $KEY"

# Create a task from the JSON request body in task.json.
curl --connect-timeout 5 --max-time 15 -i -X POST "$HOST/api/v1/tasks/mail/create" \
  -H "Authorization: Token $KEY" -H "Content-Type: application/json" -d @task.json

# Collect the results of one group. Repeat every 3 seconds; each result is delivered once.
curl --connect-timeout 5 --max-time 15 -i "$HOST/api/v1/tasks/completed?group=signup-bot" -H "Authorization: Token $KEY"

Node.js

// Node.js 18+, no dependencies.  Run: MAILMAN_TASKS_KEY=<key> node client.mjs
const HOST = "https://tasks.getmailman.app";
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, Math.ceil(ms)));

class MailmanError extends Error {
  constructor(status, body, retryAfter) {
    super(`Tasks API ${status}: ${body}`);
    this.status = status; // HTTP status
    this.retryAfter = retryAfter; // seconds from Retry-After, 0 if absent
  }
}

class MailmanTasks {
  constructor(key, { group = "default", host = HOST } = {}) {
    Object.assign(this, { key, group, host });
    this.results = new Map(); // retained even if a later request fails
    this.expires = new Map();
    this.nextPoll = performance.now() + 3000;
    this.pollGap = 3000;
  }

  // Creates retry only explicit 429/503 rejections; never ambiguous transport failures.
  async request(method, path, body, deadline = Infinity) {
    const poll = path.startsWith("/api/v1/tasks/completed");
    let retryDelay = 0;
    for (let attempt = 0; ; attempt++) {
      const now = performance.now();
      const delay = Math.max(retryDelay, poll ? this.nextPoll - now : 0, 0);
      if (delay >= deadline - now) {
        await sleep(Math.max(0, deadline - now));
        throw new Error("task lost: deadline expired"); // no room for another paced attempt
      }
      await sleep(delay);
      if (performance.now() >= deadline) throw new Error("task lost: deadline expired");
      let res;
      try {
        res = await fetch(this.host + path, {
          method,
          headers: { Authorization: `Token ${this.key}`, "Content-Type": "application/json" },
          body: body === undefined ? undefined : JSON.stringify(body),
          signal: AbortSignal.timeout(Math.max(1, Math.ceil(poll ? 15_000 : Math.min(15_000, deadline - performance.now())))),
        });
      } catch (err) {
        if (method !== "GET" || attempt >= 5) throw err;
        retryDelay = 2 ** attempt * 1000;
        continue;
      } finally {
        if (poll) {
          this.pollGap = 3000; // the full-page allowance applies to one attempt only
          this.nextPoll = performance.now() + this.pollGap;
        }
      }
      if (res.ok) return res.json();
      const text = await res.text();
      const retryAfter = Number(res.headers.get("retry-after")) || 0;
      const retry = (res.status === 503 || (method === "GET" && res.status === 408)) || (res.status === 429 && retryAfter > 0);
      if (!retry || attempt >= 5) throw new MailmanError(res.status, text, retryAfter);
      retryDelay = Math.max(retryAfter, 2 ** attempt) * 1000;
    }
  }

  verify() {
    return this.request("GET", "/api/v1/user/verify");
  }

  // Creates a task in this client's group and returns its id.
  async createTask(task) {
    const { id } = await this.request("POST", "/api/v1/tasks/mail/create", { ...task, group: this.group });
    return id;
  }

  // Keep one client per group; serialize waits and polls through it.
  // Every page is stored before returning, including ids no caller requested.
  // The deadline gates sends/retries, never the body of a poll already sent.
  async completed(deadline = Infinity) {
    const { tasks } = await this.request("GET", `/api/v1/tasks/completed?group=${encodeURIComponent(this.group)}`, undefined, deadline);
    this.pollGap = tasks.length === 100 ? 1000 : 3000;
    this.nextPoll = performance.now() + this.pollGap;
    for (const task of tasks) {
      this.results.set(task.id, task);
      this.expires.set(task.id, performance.now() + 3600_000);
    }
    this.prune();
    return tasks;
  }

  prune() {
    for (const [id, expiry] of this.expires) if (expiry <= performance.now()) {
      this.results.delete(id);
      this.expires.delete(id);
    }
  }

  async waitForResults(ids, timeoutSeconds = 3600) {
    const deadline = performance.now() + (timeoutSeconds + 15) * 1000;
    this.prune();
    while (ids.some((id) => !this.results.has(id)) && performance.now() < deadline) {
      try {
        await this.completed(deadline);
      } catch (err) {
        if (performance.now() >= deadline) break;
        if (err.status !== 401) throw err; // collected results remain in mail.results
        this.nextPoll = performance.now() + 10_000;
      }
    }
    return new Map(ids.filter((id) => this.results.has(id)).map((id) => [id, this.results.get(id)]));
  }
}

// ---- Usage ----
const mail = new MailmanTasks(process.env.MAILMAN_TASKS_KEY, { group: "signup-bot" });

try {
  await mail.verify();
} catch (err) {
  if (err.status !== 401) throw err;
  console.error("The key was rejected, or Mailman isn't open on the user's computer.");
  process.exit(1);
}

const timeout = 120;
const id = await mail.createTask({
  email: "alias@example.com",
  receivedAt: Math.floor(Date.now() / 1000),
  timeout,
  mailFilters: [
    { target: "from", comparator: "includes", value: "noreply@example.com" },
    { target: "subject", comparator: "includes", value: "verification code" },
  ],
  mailElements: [{ name: "otpCode", target: "body", regex: "([0-9]{6})" }],
});

const result = (await mail.waitForResults([id], timeout)).get(id);
if (result?.status === "success") console.log("code:", result.results.otpCode ?? "no code found");
else console.log("no code:", result ?? "lost");

console.log("a second poll returns", (await mail.completed()).length, "tasks"); // 0: delivered once

Python

# Python 3.8+, standard library only.  Run: MAILMAN_TASKS_KEY=<key> python client.py
import http.client
import queue
import socket
import threading
import json
import os
import sys
import time
from urllib.parse import quote, urlsplit

HOST = "https://tasks.getmailman.app"


def fetch_text(method, url, data=None, key="", deadline=float("inf")):
    """A 15s total request budget, also bounded by the caller's absolute deadline.

    The caller owns the clock, including DNS/connect, headers and a trickling body.
    Shutdown interrupts active socket I/O; a late DNS lookup cannot send a request.
    """
    deadline = min(deadline, time.monotonic() + 15)
    parts = urlsplit(url)
    connection = (http.client.HTTPSConnection if parts.scheme == "https" else http.client.HTTPConnection)(
        parts.hostname, parts.port, timeout=max(0.001, deadline - time.monotonic()))
    done, cancelled, sockets = queue.Queue(maxsize=1), threading.Event(), []

    def exchange():
        try:
            connection.connect()
            sock = connection.sock
            sockets.append(sock)
            if cancelled.is_set() or time.monotonic() >= deadline:
                raise TimeoutError("request deadline expired")
            connection.request(method, parts.path + ("?" + parts.query if parts.query else ""),
                               body=data, headers={"Authorization": f"Token {key}",
                                                   "Content-Type": "application/json"})
            response = connection.getresponse()
            with response:
                text = response.read().decode("utf-8", errors="replace")
                done.put((response.status, response.headers, text))
        except Exception as err:
            done.put(err)
        finally:
            connection.close()

    # Daemon only matters if the system resolver itself stalls; Python cannot cancel DNS.
    threading.Thread(target=exchange, daemon=True).start()
    try:
        answer = done.get(timeout=max(0, deadline - time.monotonic()))
    except queue.Empty:
        raise TimeoutError("request deadline expired") from None
    finally:
        cancelled.set()
        for sock in sockets:
            try:
                sock.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
    if time.monotonic() >= deadline:
        raise TimeoutError("request deadline expired")
    if isinstance(answer, Exception):
        raise answer
    return answer


class MailmanError(Exception):
    def __init__(self, status, body, retry_after=0):
        super().__init__(f"Tasks API {status}: {body}")
        self.status = status  # HTTP status
        self.retry_after = retry_after  # seconds from Retry-After, 0 if absent


class MailmanTasks:
    def __init__(self, key, group="default", host=HOST):
        self.key, self.group, self.host = key, group, host
        self.results, self.expires = {}, {}  # survive later failures
        self.next_poll, self.poll_gap = time.monotonic() + 3, 3

    def request(self, method, path, body=None, deadline=float("inf")):
        """Creates retry only explicit 429/503 rejections, never transport failures."""
        poll = path.startswith("/api/v1/tasks/completed")
        retry_delay = 0
        data = None if body is None else json.dumps(body).encode()
        for attempt in range(6):
            now = time.monotonic()
            delay = max(retry_delay, self.next_poll - now if poll else 0, 0)
            time.sleep(min(delay, max(0, deadline - now)))
            if time.monotonic() >= deadline:
                raise TimeoutError("task lost: deadline expired")
            try:
                status, headers, text = fetch_text(method, self.host + path, data,
                                                   key=self.key, deadline=float("inf") if poll else deadline)
                if status == 200:
                    return json.loads(text)
                header = headers.get("Retry-After", "")
                retry_after = int(header) if header.isdigit() else 0
                retry = (status == 503 or (method == "GET" and status == 408)) or (status == 429 and retry_after > 0)
                if not retry or attempt == 5:
                    raise MailmanError(status, text, retry_after)
                retry_delay = max(retry_after, 2 ** attempt)
            except (OSError, http.client.HTTPException):
                if method != "GET" or attempt == 5:
                    raise
                retry_delay = 2 ** attempt

            finally:
                if poll:
                    self.poll_gap = 3  # a full page permits only the very next attempt at 1s
                    self.next_poll = time.monotonic() + self.poll_gap

    def verify(self):
        return self.request("GET", "/api/v1/user/verify")

    def create_task(self, task):
        """Creates a task in this client's group and returns its id."""
        return self.request("POST", "/api/v1/tasks/mail/create", {**task, "group": self.group})["id"]

    # Keep one client per group; serialize waits and polls through it.
    # The deadline gates sends/retries, never the body of a poll already sent.
    def completed(self, deadline=float("inf")):
        tasks = self.request("GET", f"/api/v1/tasks/completed?group={quote(self.group, safe='')}", deadline=deadline)["tasks"]
        self.poll_gap = 1 if len(tasks) == 100 else 3
        self.next_poll = time.monotonic() + self.poll_gap
        for task in tasks:  # store even ids nobody requested
            self.results[task["id"]] = task
            self.expires[task["id"]] = time.monotonic() + 3600
        self.prune()
        return tasks

    def prune(self):
        for task_id, expiry in list(self.expires.items()):
            if expiry <= time.monotonic():
                self.results.pop(task_id, None)
                del self.expires[task_id]

    def wait_for_results(self, ids, timeout_seconds=3600):
        deadline = time.monotonic() + timeout_seconds + 15
        self.prune()
        while any(i not in self.results for i in ids) and time.monotonic() < deadline:
            try:
                self.completed(deadline)
            except Exception as err:
                if time.monotonic() >= deadline:
                    break
                if not isinstance(err, MailmanError) or err.status != 401:
                    raise  # collected results remain in mail.results
                self.next_poll = time.monotonic() + 10
        return {i: self.results[i] for i in ids if i in self.results}


# ---- Usage ----
if __name__ == "__main__":
    mail = MailmanTasks(os.environ["MAILMAN_TASKS_KEY"], group="signup-bot")

    try:
        mail.verify()
    except MailmanError as err:
        if err.status != 401:
            raise
        sys.exit("The key was rejected, or Mailman isn't open on the user's computer.")

    timeout = 120
    task_id = mail.create_task({
        "email": "alias@example.com",
        "receivedAt": int(time.time()),
        "timeout": timeout,
        "mailFilters": [
            {"target": "from", "comparator": "includes", "value": "noreply@example.com"},
            {"target": "subject", "comparator": "includes", "value": "verification code"},
        ],
        "mailElements": [{"name": "otpCode", "target": "body", "regex": "([0-9]{6})"}],
    })

    result = mail.wait_for_results([task_id], timeout).get(task_id)
    if result and result["status"] == "success":
        print("code:", result["results"].get("otpCode", "no code found"))
    else:
        print("no code:", result or "lost")

    print("a second poll returns", len(mail.completed()), "tasks")  # 0: delivered once

Go

// Go 1.21+, standard library only.  Run: MAILMAN_TASKS_KEY=<key> go run client.go
package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"
)

const host = "https://tasks.getmailman.app"

type MailFilter struct {
	Target          string   `json:"target"`
	Comparator      string   `json:"comparator"`
	Value           string   `json:"value"`
	OrValues        []string `json:"orValues,omitempty"`
	RegexReplaceAll string   `json:"regexReplaceAll,omitempty"`
	Optional        bool     `json:"optional,omitempty"`
}

type MailElement struct {
	Name        string `json:"name"`
	Target      string `json:"target"`
	Selector    string `json:"selector,omitempty"`
	Regex       string `json:"regex,omitempty"`
	RegexSource string `json:"regexSource,omitempty"`
	Formatter   string `json:"formatter,omitempty"`
}

type MailTask struct {
	TemplateID   string        `json:"templateId,omitempty"`
	Group        string        `json:"group,omitempty"`
	Email        string        `json:"email"`
	ReceivedAt   int64         `json:"receivedAt"`
	Timeout      int           `json:"timeout,omitempty"`
	MailFilters  []MailFilter  `json:"mailFilters"`
	MailElements []MailElement `json:"mailElements"`
}

type TaskResult struct {
	ID      string            `json:"id"`
	Status  string            `json:"status"` // "success", "error" or "timeout"
	Results map[string]string `json:"results"`
}

// APIError is any answer other than 200 that was not retried away.
type APIError struct {
	Status     int
	Body       string
	RetryAfter time.Duration // from Retry-After, 0 if absent
}

func (e *APIError) Error() string { return fmt.Sprintf("tasks api %d: %s", e.Status, e.Body) }

type Client struct {
	Key, Group, Host string
	HTTP             *http.Client
	Results          map[string]TaskResult // all pages survive later failures
	expires          map[string]time.Time
	nextPoll         time.Time
	pollGap          time.Duration
}

// Disable connection reuse so net/http cannot transparently replay a poll.
func NewClient(key, group string) *Client {
	return &Client{Key: key, Group: group, Host: host, HTTP: &http.Client{Timeout: 15 * time.Second, Transport: &http.Transport{DisableKeepAlives: true}},
		Results: map[string]TaskResult{}, expires: map[string]time.Time{},
		nextPoll: time.Now().Add(3 * time.Second), pollGap: 3 * time.Second}
}

// Creates retry only explicit 429/503 rejections; never ambiguous transport failures.
func (c *Client) do(ctx context.Context, method, path string, in, out any) error {
	var payload []byte
	if in != nil {
		payload, _ = json.Marshal(in)
	}
	poll := strings.HasPrefix(path, "/api/v1/tasks/completed")
	retryDelay := time.Duration(0)
	for attempt := 0; ; attempt++ {
		delay := retryDelay
		if poll && time.Until(c.nextPoll) > delay {
			delay = time.Until(c.nextPoll)
		}
		timer := time.NewTimer(max(delay, 0))
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
		if err := ctx.Err(); err != nil {
			return err
		}
		// Once sent, a poll runs to HTTP.Timeout, independently of the wait.
		requestCtx := ctx
		if poll {
			requestCtx = context.Background()
		}
		req, err := http.NewRequestWithContext(requestCtx, method, c.Host+path, bytes.NewReader(payload))
		if err != nil {
			return err
		}
		req.Header.Set("Authorization", "Token "+c.Key)
		req.Header.Set("Content-Type", "application/json")
		res, err := c.HTTP.Do(req)
		if poll {
			c.pollGap = 3 * time.Second // consume the one-attempt full-page allowance
			c.nextPoll = time.Now().Add(c.pollGap)
		}
		if err != nil {
			if method != http.MethodGet || attempt >= 5 {
				return err
			}
			retryDelay = time.Duration(1<<attempt) * time.Second
			continue
		}
		data, readErr := io.ReadAll(res.Body)
		res.Body.Close()
		if readErr != nil {
			return readErr
		}
		if res.StatusCode == http.StatusOK {
			if out == nil {
				return nil
			}
			return json.Unmarshal(data, out)
		}
		seconds, _ := strconv.Atoi(res.Header.Get("Retry-After"))
		retryAfter := time.Duration(seconds) * time.Second
		retry := (method == http.MethodGet && res.StatusCode == 408) || res.StatusCode == 503 || (res.StatusCode == 429 && seconds > 0)
		if !retry || attempt >= 5 {
			return &APIError{res.StatusCode, string(data), retryAfter}
		}
		retryDelay = time.Duration(1<<attempt) * time.Second
		if retryAfter > retryDelay {
			retryDelay = retryAfter
		}
	}
}

func (c *Client) Verify() error {
	return c.do(context.Background(), "GET", "/api/v1/user/verify", nil, nil)
}

// CreateTask creates a task in this client's group and returns its id.
func (c *Client) CreateTask(task MailTask) (string, error) {
	task.Group = c.Group
	var created TaskResult
	err := c.do(context.Background(), "POST", "/api/v1/tasks/mail/create", task, &created)
	return created.ID, err
}

// Keep one client per group; serialize waits and polls through it. Creates may run concurrently.
// Completed stores every result, including ids nobody requested.
func (c *Client) Completed() ([]TaskResult, error) {
	return c.completed(context.Background())
}

func (c *Client) completed(ctx context.Context) ([]TaskResult, error) {
	var page struct {
		Tasks []TaskResult `json:"tasks"`
	}
	err := c.do(ctx, "GET", "/api/v1/tasks/completed?group="+url.QueryEscape(c.Group), nil, &page)
	if err != nil {
		return nil, err
	}
	c.pollGap = 3 * time.Second
	if len(page.Tasks) == 100 {
		c.pollGap = time.Second
	}
	c.nextPoll = time.Now().Add(c.pollGap)
	for _, t := range page.Tasks {
		c.Results[t.ID] = t
		c.expires[t.ID] = time.Now().Add(time.Hour)
	}
	c.prune()
	return page.Tasks, nil
}

func (c *Client) prune() {
	for id, expiry := range c.expires {
		if !time.Now().Before(expiry) {
			delete(c.Results, id)
			delete(c.expires, id)
		}
	}
}

func (c *Client) WaitForResults(ids []string, timeoutSeconds int) (map[string]TaskResult, error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutSeconds+15)*time.Second)
	defer cancel()
	c.prune()
	collected := func() map[string]TaskResult {
		out := map[string]TaskResult{}
		for _, id := range ids {
			if t, ok := c.Results[id]; ok {
				out[id] = t
			}
		}
		return out
	}
	for len(collected()) < len(ids) && ctx.Err() == nil {
		_, err := c.completed(ctx)
		if ctx.Err() != nil {
			break
		}
		var apiErr *APIError
		if errors.As(err, &apiErr) && apiErr.Status == http.StatusUnauthorized {
			c.nextPoll = time.Now().Add(10 * time.Second)
			continue
		}
		if err != nil {
			return collected(), err
		} // all pages also remain in c.Results
	}
	return collected(), nil // absent ids are lost
}

func main() {
	mail := NewClient(os.Getenv("MAILMAN_TASKS_KEY"), "signup-bot")

	err := mail.Verify()
	var apiErr *APIError
	if errors.As(err, &apiErr) && apiErr.Status == http.StatusUnauthorized {
		fmt.Fprintln(os.Stderr, "The key was rejected, or Mailman isn't open on the user's computer.")
		os.Exit(1)
	}
	if err != nil {
		panic(err)
	}

	const timeout = 120
	id, err := mail.CreateTask(MailTask{
		Email:      "alias@example.com",
		ReceivedAt: time.Now().Unix(),
		Timeout:    timeout,
		MailFilters: []MailFilter{
			{Target: "from", Comparator: "includes", Value: "noreply@example.com"},
			{Target: "subject", Comparator: "includes", Value: "verification code"},
		},
		MailElements: []MailElement{{Name: "otpCode", Target: "body", Regex: "([0-9]{6})"}},
	})
	if err != nil {
		panic(err)
	}

	results, err := mail.WaitForResults([]string{id}, timeout)
	// Handle the partial results before reporting a later polling failure.
	if result, ok := results[id]; ok && result.Status == "success" {
		if code, found := result.Results["otpCode"]; found {
			fmt.Println("code:", code)
		} else {
			fmt.Println("no code found")
		}
	} else {
		fmt.Println("no code:", results[id], "(missing means lost)")
	}

	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		return
	}
	left, _ := mail.Completed()
	fmt.Println("a second poll returns", len(left), "tasks") // 0: delivered once
}