MongoDB updateOne upsert with $setOnInsert for Sume job_id

Create a unique index on job_id, then call updateOne with upsert: true and $setOnInsert. The result tells you which delivery you are looking at: upsertedCount is 1 for the first delivery and 0 for a retry. Only start downstream work when it is 1.

This follows the rule in Sume's webhook docs: job_id is your idempotency key, because a failed or slow attempt is retried up to 10 times and a Redeliver can repeat a terminal event.

The write

$setOnInsert writes its fields only when the upsert creates the document, so a retry never rewrites the stored event. $inc runs every time and gives you a delivery counter, which is useful when you debug a flaky endpoint.

import { MongoClient } from "mongodb";
const client = new MongoClient(process.env.MONGODB_URI ?? "mongodb://localhost:27017");
const jobs = client.db("app").collection("sume_jobs");
await jobs.createIndex({ job_id: 1 }, { unique: true });

export async function record(event) {
  const run = () => jobs.updateOne(
    { job_id: event.job_id },
    {
      $setOnInsert: { event: event.event, status: event.status, payload: event, first_seen: new Date() },
      $inc: { deliveries: 1 },
    },
    { upsert: true },
  );
  try {
    return (await run()).upsertedCount === 1;
  } catch (err) {
    if (err.code !== 11000) throw err; // lost a race on the unique index
    await run();
    return false;
  }
}

const first = await record({ job_id: "job_demo", event: "job.completed", status: "OK" });
const again = await record({ job_id: "job_demo", event: "job.completed", status: "OK" });
console.log({ first, again });
await jobs.deleteOne({ job_id: "job_demo" });
await client.close();

Reading the result

Run the file and it prints { first: true, again: false }. In a handler, verify the signature before you call record, and enqueue your follow-up work only when it returns true.

Caveats

• A crash between the upsert and your enqueue loses the job. Write the follow-up work in the same document, for example a needs_download: true flag, and let a sweeper pick it up.
• Concurrent upserts on an indexed field can raise duplicate key error 11000 even with upsert: true. The sample retries once, which then matches the existing document.
• Keep the status_url poll for jobs that never show up in the collection. Sume delivers terminal events only and has no progress events.

Why the key is job_id and nothing else

A document store tempts people to key on the whole event body. Do not. Two deliveries of one job carry the same job_id but a different timestamp header, and a Redeliver carries a fresh timestamp and signature on the same terminal event. The only stable identity is job_id, which is why it sits in the filter of the upsert and in the unique index.

Failed and canceled jobs use the same envelope with status: "ERROR" and an error object, so the same record function handles all three terminal events. Store the whole event in payload and decide what to do with it in the code that reads the collection.

One document per job is the whole story

Sume sends terminal events only: job.completed, job.failed and job.canceled. There are no progress or partial deliveries, so one document per job is the complete history you will receive. If you want a timeline, read the job events from the API rather than trying to build it from webhooks.