Signatures and payloads

const fs = require("node:fs");
const fsp = require("node:fs/promises");
const http = require("node:http");
const path = require("node:path");

const INBOX = path.resolve("inbox"); // recorded, not yet processed
const TMP = path.resolve("inbox-tmp"); // records being written; same disk as INBOX
const DONE = path.resolve("processed"); // processed
for (const dir of [INBOX, TMP, DONE]) fs.mkdirSync(dir, { recursive: true });

const RETRY_MS = 30_000;

async function syncDir(dir) {
  const handle = await fsp.open(dir, "r");
  try {
    await handle.sync();
  } finally {
    await handle.close();
  }
}

// Record the raw body as inbox/<id>.json. Only a complete file ever appears
// there: write and flush a temporary file, then rename it into place.
async function record(id, rawBody) {
  const name = `${id}.json`;
  if (fs.existsSync(path.join(INBOX, name))) {
    await syncDir(INBOX); // already recorded; make sure it is on disk
    return false;
  }
  if (fs.existsSync(path.join(DONE, name))) return false; // already processed

  const tmp = path.join(TMP, `${name}.${process.pid}.${Date.now()}`);
  try {
    const file = await fsp.open(tmp, "wx");
    try {
      await file.writeFile(rawBody);
      await file.sync();
    } finally {
      await file.close();
    }
    await fsp.rename(tmp, path.join(INBOX, name));
  } catch (err) {
    await fsp.rm(tmp, { force: true });
    throw err;
  }
  await syncDir(INBOX);
  return true;
}

// Your work goes here. It can run more than once for the same event, so make
// its effect idempotent, keyed by event.id.
async function handle(event) {
  console.log("handling", event.type, event.id);
}

// Process every recorded event, one at a time. An event leaves the inbox only
// after handle() has finished; one that fails stays and is retried later.
let running = null;
let again = false;
function processInbox() {
  if (running) {
    again = true;
    return running;
  }
  running = (async () => {
    do {
      again = false;
      for (const name of await fsp.readdir(INBOX)) {
        try {
          await handle(JSON.parse(await fsp.readFile(path.join(INBOX, name), "utf8")));
          await fsp.rename(path.join(INBOX, name), path.join(DONE, name));
        } catch (err) {
          console.error("processing failed, will retry", name, err);
        }
      }
    } while (again);
  })().finally(() => {
    running = null;
  });
  return running;
}

http
  .createServer((req, res) => {
    const chunks = [];
    req.on("data", (chunk) => chunks.push(chunk));
    req.on("end", async () => {
      const rawBody = Buffer.concat(chunks);

      // 1. Verify.
      if (!verifyWebhook(process.env.OXARCHIVE_WEBHOOK_SECRET, req.headers["0xa-signature"], rawBody)) {
        res.writeHead(401).end();
        return;
      }
      const event = JSON.parse(rawBody.toString("utf8"));
      if (!/^[0-9a-f-]{36}$/i.test(event.id)) {
        res.writeHead(400).end();
        return;
      }

      // 2. Record, once per event id.
      try {
        await record(event.id, rawBody);
      } catch (err) {
        console.error("not recorded, the delivery will be retried", event.id, err);
        res.writeHead(500).end();
        return;
      }

      // 3. Acknowledge.
      res.writeHead(204).end();

      // 4. Process off the request.
      processInbox();
    });
  })
  .listen(8080);

processInbox(); // anything recorded before a restart
setInterval(processInbox, RETRY_MS); // retry processing that failed