Simple server with synchronization of request processing

server.js

/**
 * WARNING: don't use the code in production, it only works for a single process instance and does not work in cluster mode.
 */

const express = require('express');
const app = express();

let nextRequestId = 1; // request counter
let nextPaymentId = 1; // payment counter

// POST to /payments creates a new empty resource
app.post('/payments', (req, res) => {
  const paymentId = nextPaymentId++;
  const context = `request(post) #${nextRequestId++}`;
  handle(() => createPayment(context, paymentId), res);
});

// PUT to a payment resource charges the user
// app.put('/payments/:id', (req, res) => {
//   const context = `request(put) #${nextRequestId++}`;
//   const paymentId = req.params.id;
//   handle(() => conductPayment(context, paymentId), res);
// });

// PUT to a payment resource charges the user
// with locks
app.put('/payments/:id', (req, res) => {
  const context = `request(put) #${nextRequestId++}`;
  const paymentId = req.params.id;
  handleWithLock(
    context,
    paymentId,
    () => conductPayment(context, paymentId),
    res
  );
});

app.listen(3000, () => console.log('Example app listening on port 3000!'));

// simple payments database
const _payments = {}; // { id: string, state: EMPTY | PROCESSING | PAID }

async function getPayment(context, paymentId) {
  console.log(timestamp(), context, `Payment ${paymentId} retrieved`);
  return _payments[paymentId];
}

async function createPayment(context, paymentId) {
  console.log(timestamp(), context, `Payment ${paymentId} created`);
  _payments[paymentId] = {
    id: paymentId,
    state: 'EMPTY',
  };
  return _payments[paymentId];
}

async function processPayment(context, paymentId) {
  console.log(timestamp(), context, `Payment ${paymentId} processing started`);
  _payments[paymentId].state = 'PROCESSING';
  // payment processing is a lengthy operation
  await new Promise(resolve => {
    setTimeout(() => {
      console.log(timestamp(), context, `Procesed payment ${paymentId}`);
      _payments[paymentId].state = 'PAID';
      resolve(_payments[paymentId]);
    }, 3000);
  });
}

async function conductPayment(context, paymentId) {
  const payment = await getPayment(context, paymentId);
  if (!payment) {
    throw new Error('Payment does not exist');
  }
  if (payment.state === 'PROCESSING') {
    // TODO improve by waiting for the state change
    throw new Error('Payment is in progress. Try again later.');
  }
  if (payment.state === 'PAID') {
    return payment;
  }
  if (payment.state === 'EMPTY') {
    await processPayment(context, paymentId);
  }
  throw new Error('Payment is in bad state');
}

async function handle(fn, res) {
  try {
    const result = await fn();
    if (result) {
      return res.status(200).json(result);
    }
    res.status(204).end();
  } catch (err) {
    res.status(409).json({
      error: err.message,
    });
  }
}

async function handleWithLock(context, lockId, fn, res) {
  let lockState;
  try {
    lockState = await lock(context, lockId);
    if (lockState === 'locked') throw new Error('Resource is locked.');
    const result = await fn();
    if (result) {
      return res.status(200).json(result);
    }
    res.status(204).end();
  } catch (err) {
    res.status(409).json({
      error: err.message,
    });
  } finally {
    if (lockState === 'acquired') {
      await unlock(context, lockId);
    }
  }
}

function timestamp() {
  return new Date().toUTCString();
}

// simple locking (don't use in production)
const _locks = {};

async function lock(context, lockId) /* : 'locked' | 'acquired' */ {
  if (!_locks[lockId]) {
    _locks[lockId] = context;
    console.log(timestamp(), context, 'lock acquired');
    return 'acquired';
  } else {
    console.log(
      timestamp(),
      context,
      `failed to lock the payment: payment is already locked by ${
        _locks[lockId]
      }`
    );
    return 'locked';
  }
}

async function unlock(context, lockId) {
  console.log(timestamp(), context, `payment is unlocked`);
  delete _locks[lockId];
}

In a cluster setup, I'm assuming you folks use queuing?