Retries, ordering and deduplication

# The ID to record: an event keeps it across retries and resends.
curl https://pay-api.adpromax.com/v1/events/evt_01k7d6b4c9x3z7v2n5m8q1w4ry \
  -u "$APX_SECRET_KEY:"

// PostgreSQL with node-postgres:
// CREATE TABLE handled_events (id text PRIMARY KEY);
export async function handleOnce(pool, event, work) {
  const client = await pool.connect()
  try {
    await client.query('BEGIN')
    const inserted = await client.query(
      'INSERT INTO handled_events (id) VALUES ($1) ON CONFLICT DO NOTHING',
      [event.id],
    )
    if (inserted.rowCount === 1)
      await work(client, event)
    await client.query('COMMIT')
  }
  catch (error) {
    await client.query('ROLLBACK')
    throw error // answer 500: the delivery will be retried
  }
  finally {
    client.release()
  }
}

// PostgreSQL with node-postgres (its Pool and PoolClient fit these shapes):
// CREATE TABLE handled_events (id text PRIMARY KEY);
interface Client {
  query: (sql: string, values?: unknown[]) => Promise<{ rowCount: number | null }>
  release: () => void
}

interface Pool {
  connect: () => Promise<Client>
}

interface WebhookEvent {
  id: string
  type: string
}

export async function handleOnce(
  pool: Pool,
  event: WebhookEvent,
  work: (client: Client, event: WebhookEvent) => Promise<void>,
): Promise<void> {
  const client = await pool.connect()
  try {
    await client.query('BEGIN')
    const inserted = await client.query(
      'INSERT INTO handled_events (id) VALUES ($1) ON CONFLICT DO NOTHING',
      [event.id],
    )
    if (inserted.rowCount === 1)
      await work(client, event)
    await client.query('COMMIT')
  }
  catch (error) {
    await client.query('ROLLBACK')
    throw error // answer 500: the delivery will be retried
  }
  finally {
    client.release()
  }
}

# SQLite:
# CREATE TABLE handled_events (id TEXT PRIMARY KEY);
def handle_once(db, event, work):
    with db:  # one transaction: committed, or rolled back on error
        inserted = db.execute(
            "INSERT OR IGNORE INTO handled_events (id) VALUES (?)",
            (event["id"],),
        )
        if inserted.rowcount == 1:
            work(db, event)

<?php
// MySQL with PDO:
// CREATE TABLE handled_events (id VARCHAR(64) PRIMARY KEY);
function handle_once(PDO $db, array $event, callable $work): void
{
    $db->beginTransaction();
    try {
        $insert = $db->prepare(
            'INSERT IGNORE INTO handled_events (id) VALUES (?)'
        );
        $insert->execute([$event['id']]);
        if ($insert->rowCount() === 1) {
            $work($db, $event);
        }
        $db->commit();
    } catch (Throwable $error) {
        $db->rollBack();
        throw $error; // answer 500: the delivery will be retried
    }
}

# PostgreSQL with the pg gem:
# CREATE TABLE handled_events (id text PRIMARY KEY);
def handle_once(db, event)
  # One transaction: committed, or rolled back when the block raises
  # (answer 500 then: the delivery will be retried).
  db.transaction do |tx|
    inserted = tx.exec_params(
      'INSERT INTO handled_events (id) VALUES ($1) ON CONFLICT DO NOTHING',
      [event['id']]
    )
    yield tx, event if inserted.cmd_tuples == 1
  end
end

package main

import (
	"context"
	"database/sql"
)

type Event struct {
	ID   string `json:"id"`
	Type string `json:"type"`
}

// PostgreSQL with database/sql:
// CREATE TABLE handled_events (id text PRIMARY KEY);
func HandleOnce(ctx context.Context, db *sql.DB, event Event,
	work func(*sql.Tx, Event) error) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback() // does nothing once committed
	inserted, err := tx.ExecContext(ctx,
		"INSERT INTO handled_events (id) VALUES ($1) ON CONFLICT DO NOTHING",
		event.ID)
	if err != nil {
		return err
	}
	if rows, _ := inserted.RowsAffected(); rows == 1 {
		if err := work(tx, event); err != nil {
			return err // answer 500: the delivery will be retried
		}
	}
	return tx.Commit()
}

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import javax.sql.DataSource;

public class HandleOnce {
    public interface Work {
        void run(Connection db, String eventId) throws SQLException;
    }

    // PostgreSQL with JDBC:
    // CREATE TABLE handled_events (id text PRIMARY KEY);
    public static void handleOnce(DataSource pool, String eventId, Work work)
            throws SQLException {
        try (Connection db = pool.getConnection()) {
            db.setAutoCommit(false);
            try (PreparedStatement insert = db.prepareStatement(
                    "INSERT INTO handled_events (id) VALUES (?)"
                        + " ON CONFLICT DO NOTHING")) {
                insert.setString(1, eventId);
                if (insert.executeUpdate() == 1) {
                    work.run(db, eventId);
                }
                db.commit();
            } catch (SQLException | RuntimeException error) {
                db.rollback();
                throw error; // answer 500: the delivery will be retried
            }
        }
    }
}

using System.Data.Common;

public static class HandleOnce
{
    // PostgreSQL with ADO.NET (an open NpgsqlConnection):
    // CREATE TABLE handled_events (id text PRIMARY KEY);
    public static async Task Run(
        DbConnection db, string eventId, Func<DbTransaction, Task> work)
    {
        // One transaction: rolled back when disposed without a commit
        // (answer 500 then: the delivery will be retried).
        await using var tx = await db.BeginTransactionAsync();
        await using var insert = db.CreateCommand();
        insert.Transaction = tx;
        insert.CommandText =
            "INSERT INTO handled_events (id) VALUES (@id) ON CONFLICT DO NOTHING";
        var id = insert.CreateParameter();
        id.ParameterName = "id";
        id.Value = eventId;
        insert.Parameters.Add(id);
        if (await insert.ExecuteNonQueryAsync() == 1)
            await work(tx);
        await tx.CommitAsync();
    }
}

// Cargo.toml: postgres = "0.19"
// CREATE TABLE handled_events (id text PRIMARY KEY);
use postgres::{Client, Error, Transaction};

pub fn handle_once(
    db: &mut Client,
    event_id: &str,
    work: impl FnOnce(&mut Transaction) -> Result<(), Error>,
) -> Result<(), Error> {
    // One transaction: rolled back when dropped without a commit
    // (answer 500 then: the delivery will be retried).
    let mut tx = db.transaction()?;
    let inserted = tx.execute(
        "INSERT INTO handled_events (id) VALUES ($1) ON CONFLICT DO NOTHING",
        &[&event_id],
    )?;
    if inserted == 1 {
        work(&mut tx)?;
    }
    tx.commit()
}

curl https://pay-api.adpromax.com/v1/payment_intents/pi_01k7d2h6w0f3r8t5y2v9n4p6qa \
  -u "$APX_SECRET_KEY:"

const id = event.data.object.id
const response = await fetch(`https://pay-api.adpromax.com/v1/payment_intents/${id}`, {
  headers: { Authorization: `Bearer ${process.env.APX_SECRET_KEY}` },
})
const paymentIntent = await response.json()
if (paymentIntent.status === 'succeeded')
  console.log('deliver', paymentIntent.id)

interface PaymentIntent {
  id: string
  status: string
}

interface PaymentIntentEvent {
  data: { object: PaymentIntent }
}

// `event`: the delivery your handler verified and parsed.
export async function deliverIfSucceeded(event: PaymentIntentEvent): Promise<void> {
  const id = event.data.object.id
  const response = await fetch(`https://pay-api.adpromax.com/v1/payment_intents/${id}`, {
    headers: { Authorization: `Bearer ${process.env.APX_SECRET_KEY}` },
  })
  const paymentIntent = (await response.json()) as PaymentIntent
  if (paymentIntent.status === 'succeeded')
    console.log('deliver', paymentIntent.id)
}

import os

import requests

payment_intent_id = event["data"]["object"]["id"]
payment_intent = requests.get(
    f"https://pay-api.adpromax.com/v1/payment_intents/{payment_intent_id}",
    auth=(os.environ["APX_SECRET_KEY"], ""),
).json()
if payment_intent["status"] == "succeeded":
    print("deliver", payment_intent["id"])

<?php
$id = $event['data']['object']['id'];
$ch = curl_init('https://pay-api.adpromax.com/v1/payment_intents/' . urlencode($id));
curl_setopt($ch, CURLOPT_USERPWD, getenv('APX_SECRET_KEY') . ':');
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$paymentIntent = json_decode(curl_exec($ch), true);
if ($paymentIntent['status'] === 'succeeded') {
    error_log('deliver ' . $paymentIntent['id']);
}

require 'json'
require 'net/http'

id = event['data']['object']['id']
uri = URI("https://pay-api.adpromax.com/v1/payment_intents/#{id}")
request = Net::HTTP::Get.new(uri)
request.basic_auth(ENV.fetch('APX_SECRET_KEY'), '')
response = Net::HTTP.start(uri.hostname, uri.port, use_ssl: uri.scheme == 'https') do |http|
  http.request(request)
end
payment_intent = JSON.parse(response.body)
puts "deliver #{payment_intent['id']}" if payment_intent['status'] == 'succeeded'

package main

import (
	"encoding/json"
	"log"
	"net/http"
	"os"
)

type PaymentIntent struct {
	ID     string `json:"id"`
	Status string `json:"status"`
}

type Event struct {
	Data struct {
		Object PaymentIntent `json:"object"`
	} `json:"data"`
}

// event: the delivery your handler verified and parsed.
func deliverIfSucceeded(event Event) error {
	req, err := http.NewRequest(http.MethodGet,
		"https://pay-api.adpromax.com/v1/payment_intents/"+event.Data.Object.ID, nil)
	if err != nil {
		return err
	}
	req.SetBasicAuth(os.Getenv("APX_SECRET_KEY"), "")
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	defer res.Body.Close()

	var paymentIntent PaymentIntent
	if err := json.NewDecoder(res.Body).Decode(&paymentIntent); err != nil {
		return err
	}
	if paymentIntent.Status == "succeeded" {
		log.Println("deliver", paymentIntent.ID)
	}
	return nil
}

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.regex.Pattern;

public class CurrentState {
    // No JSON library here: a small regex reads the status.
    static final Pattern SUCCEEDED =
        Pattern.compile("\"status\"\\s*:\\s*\"succeeded\"");

    // id: the payment intent's ID, from the event's data.object.
    static void deliverIfSucceeded(String id) throws Exception {
        HttpRequest request = HttpRequest.newBuilder(
                URI.create("https://pay-api.adpromax.com/v1/payment_intents/" + id))
            .header("Authorization", "Bearer " + System.getenv("APX_SECRET_KEY"))
            .GET()
            .build();
        String paymentIntent = HttpClient.newHttpClient()
            .send(request, HttpResponse.BodyHandlers.ofString())
            .body();
        if (SUCCEEDED.matcher(paymentIntent).find()) {
            System.out.println("deliver " + id);
        }
    }
}

using System.Net.Http.Headers;
using System.Net.Http.Json;
using System.Text.Json;

var json = new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower };
using var http = new HttpClient();
http.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue(
    "Bearer", Environment.GetEnvironmentVariable("APX_SECRET_KEY"));

// id: the payment intent's ID, from the event's data.object.
async Task DeliverIfSucceeded(string id)
{
    var paymentIntent = await http.GetFromJsonAsync<PaymentIntent>(
        $"https://pay-api.adpromax.com/v1/payment_intents/{id}", json);
    if (paymentIntent?.Status == "succeeded")
        Console.WriteLine($"deliver {paymentIntent.Id}");
}

record PaymentIntent(string Id, string Status);

// Cargo.toml: reqwest = { version = "0.12", features = ["blocking", "json"] }
//             serde = { version = "1", features = ["derive"] }
use serde::Deserialize;

#[derive(Deserialize)]
struct PaymentIntent {
    id: String,
    status: String,
}

/// `id`: the payment intent's ID, from the event's `data.object`.
fn deliver_if_succeeded(id: &str) -> Result<(), Box<dyn std::error::Error>> {
    let payment_intent: PaymentIntent = reqwest::blocking::Client::new()
        .get(format!("https://pay-api.adpromax.com/v1/payment_intents/{id}"))
        .bearer_auth(std::env::var("APX_SECRET_KEY")?)
        .send()?
        .json()?;
    if payment_intent.status == "succeeded" {
        println!("deliver {}", payment_intent.id);
    }
    Ok(())
}

curl "https://pay-api.adpromax.com/v1/events?limit=100" \
  -u "$APX_SECRET_KEY:"

# the next page: older than the last event of this one
curl "https://pay-api.adpromax.com/v1/events?limit=100&starting_after=evt_01k7d6b4c9x3z7v2n5m8q1w4ry" \
  -u "$APX_SECRET_KEY:"

const headers = { Authorization: `Bearer ${process.env.APX_SECRET_KEY}` }
let startingAfter = null
let hasMore = true
while (hasMore) {
  const url = new URL('https://pay-api.adpromax.com/v1/events')
  url.searchParams.set('limit', '100')
  if (startingAfter !== null)
    url.searchParams.set('starting_after', startingAfter)
  const page = await fetch(url, { headers }).then(response => response.json())
  for (const event of page.data)
    console.log(event.id, event.type) // handle each event once
  hasMore = page.has_more && page.data.length > 0
  startingAfter = page.data.at(-1)?.id ?? null
}

interface ApiEvent {
  id: string
  type: string
}

interface EventList {
  object: 'list'
  data: ApiEvent[]
  has_more: boolean
}

const headers = { Authorization: `Bearer ${process.env.APX_SECRET_KEY}` }
let startingAfter: string | null = null
let hasMore = true
while (hasMore) {
  const url = new URL('https://pay-api.adpromax.com/v1/events')
  url.searchParams.set('limit', '100')
  if (startingAfter !== null)
    url.searchParams.set('starting_after', startingAfter)
  const response = await fetch(url, { headers })
  const page = (await response.json()) as EventList
  for (const event of page.data)
    console.log(event.id, event.type) // handle each event once
  hasMore = page.has_more && page.data.length > 0
  startingAfter = page.data.at(-1)?.id ?? null
}

import os

import requests

AUTH = (os.environ["APX_SECRET_KEY"], "")
params = {"limit": 100}
while True:
    page = requests.get(
        "https://pay-api.adpromax.com/v1/events", params=params, auth=AUTH
    ).json()
    for event in page["data"]:
        print(event["id"], event["type"])  # handle each event once
    if not page["has_more"] or not page["data"]:
        break
    params["starting_after"] = page["data"][-1]["id"]

<?php
$startingAfter = null;
do {
    $query = ['limit' => 100];
    if ($startingAfter !== null) {
        $query['starting_after'] = $startingAfter;
    }
    $ch = curl_init('https://pay-api.adpromax.com/v1/events?' . http_build_query($query));
    curl_setopt($ch, CURLOPT_USERPWD, getenv('APX_SECRET_KEY') . ':');
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
    $page = json_decode(curl_exec($ch), true);
    foreach ($page['data'] as $event) {
        echo $event['id'], ' ', $event['type'], PHP_EOL; // handle each once
    }
    $last = end($page['data']);
    $startingAfter = $last === false ? null : $last['id'];
} while ($page['has_more'] && $startingAfter !== null);

require 'json'
require 'net/http'

uri = URI('https://pay-api.adpromax.com/v1/events')
params = { limit: 100 }
loop do
  uri.query = URI.encode_www_form(params)
  request = Net::HTTP::Get.new(uri)
  request.basic_auth(ENV.fetch('APX_SECRET_KEY'), '')
  response = Net::HTTP.start(uri.hostname, uri.port, use_ssl: uri.scheme == 'https') do |http|
    http.request(request)
  end
  page = JSON.parse(response.body)
  page['data'].each do |event|
    puts "#{event['id']} #{event['type']}" # handle each event once
  end
  break if !page['has_more'] || page['data'].empty?

  params[:starting_after] = page['data'].last['id']
end

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"net/url"
	"os"
)

type Event struct {
	ID   string `json:"id"`
	Type string `json:"type"`
}

type EventList struct {
	Data    []Event `json:"data"`
	HasMore bool    `json:"has_more"`
}

func main() {
	query := url.Values{"limit": {"100"}}
	for {
		req, err := http.NewRequest(http.MethodGet,
			"https://pay-api.adpromax.com/v1/events?"+query.Encode(), nil)
		if err != nil {
			log.Fatal(err)
		}
		req.SetBasicAuth(os.Getenv("APX_SECRET_KEY"), "")
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			log.Fatal(err)
		}
		var page EventList
		err = json.NewDecoder(res.Body).Decode(&page)
		res.Body.Close()
		if err != nil {
			log.Fatal(err)
		}
		for _, event := range page.Data {
			fmt.Println(event.ID, event.Type) // handle each event once
		}
		if !page.HasMore || len(page.Data) == 0 {
			break
		}
		query.Set("starting_after", page.Data[len(page.Data)-1].ID)
	}
}

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class CatchUp {
    // No JSON library here: small regexes read each event's id and type,
    // and has_more.
    static final Pattern EVENT = Pattern.compile(
        "\"id\"\\s*:\\s*\"(evt_[^\"]+)\"[^{}]*?\"type\"\\s*:\\s*\"([^\"]+)\"");
    static final Pattern HAS_MORE = Pattern.compile("\"has_more\"\\s*:\\s*true");

    public static void main(String[] args) throws Exception {
        HttpClient http = HttpClient.newHttpClient();
        String startingAfter = null;
        boolean hasMore = true;
        while (hasMore) {
            String query = "limit=100"
                + (startingAfter == null ? "" : "&starting_after=" + startingAfter);
            HttpRequest request = HttpRequest.newBuilder(
                    URI.create("https://pay-api.adpromax.com/v1/events?" + query))
                .header("Authorization", "Bearer " + System.getenv("APX_SECRET_KEY"))
                .GET()
                .build();
            String page = http
                .send(request, HttpResponse.BodyHandlers.ofString())
                .body();
            String last = null;
            Matcher event = EVENT.matcher(page);
            while (event.find()) {
                last = event.group(1);
                // handle each event once
                System.out.println(last + " " + event.group(2));
            }
            hasMore = last != null && HAS_MORE.matcher(page).find();
            startingAfter = last;
        }
    }
}

using System.Net.Http.Headers;
using System.Net.Http.Json;
using System.Text.Json;

var json = new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower };
using var http = new HttpClient();
http.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue(
    "Bearer", Environment.GetEnvironmentVariable("APX_SECRET_KEY"));

string? startingAfter = null;
var hasMore = true;
while (hasMore)
{
    var url = "https://pay-api.adpromax.com/v1/events?limit=100"
        + (startingAfter is null ? "" : $"&starting_after={startingAfter}");
    var page = await http.GetFromJsonAsync<EventList>(url, json);
    foreach (var item in page!.Data)
        Console.WriteLine($"{item.Id} {item.Type}"); // handle each event once
    hasMore = page.HasMore && page.Data.Length > 0;
    startingAfter = page.Data.LastOrDefault()?.Id;
}

record Event(string Id, string Type);
record EventList(Event[] Data, bool HasMore);

// Cargo.toml: reqwest = { version = "0.12", features = ["blocking", "json"] }
//             serde = { version = "1", features = ["derive"] }
use serde::Deserialize;

#[derive(Deserialize)]
struct Event {
    id: String,
    #[serde(rename = "type")]
    kind: String,
}

#[derive(Deserialize)]
struct EventList {
    data: Vec<Event>,
    has_more: bool,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = reqwest::blocking::Client::new();
    let key = std::env::var("APX_SECRET_KEY")?;
    let mut starting_after: Option<String> = None;
    loop {
        let mut query = vec![("limit", "100".to_owned())];
        if let Some(id) = &starting_after {
            query.push(("starting_after", id.clone()));
        }
        let page: EventList = client
            .get("https://pay-api.adpromax.com/v1/events")
            .bearer_auth(&key)
            .query(&query)
            .send()?
            .json()?;
        for event in &page.data {
            println!("{} {}", event.id, event.kind); // handle each event once
        }
        match page.data.last() {
            Some(last) if page.has_more => starting_after = Some(last.id.clone()),
            _ => break,
        }
    }
    Ok(())
}