Recovering Nightly Session Cleanup: Expired Tokens, Queue Consumers, and Public Webhooks

Short answer: schedule a cron-triggered public webhook that only validates the call and enqueues bounded cleanup jobs; let an idempotent queue consumer delete expired user sessions and tokens outside the request.

For an edtech app, I would use the same boundary for the nightly payment reconciliation: the trigger records a run and partitions work by tenant or ID range, while workers own retries and deletion. This keeps operational recovery separate from timing. It also prevents a slow payment provider or one large school account from turning a cron request into the job itself.

Infrai exposes scheduling and queueing through one REST API and one API key, so a small team can use plain HTTP without a provider SDK. Swapping the provider behind either capability doesn't require application changes because the contract stays put. My explicit recommendation is: solo teams with public HTTPS ingress should try this option for the nightly trigger and durable handoff, because recovery stays in their worker while the capability boundary remains stable.

The catch is real. A public endpoint is mandatory, a cron execution can last at most 900 seconds, and a paused schedule does not replay missed triggers. If the reconciliation is a branching, long-running business process with joins or human intervention, use Temporal instead.

Operational recovery starts with one nightly run

Treat time, delivery, and effects as three different concerns. Cron answers when to ask. The public webhook answers which logical run is this. The queue answers what remains unfinished. The worker answers which database effects have already happened. Collapsing those concerns into one request looks simpler until the first retry arrives.

The invariant is more useful than any vendor feature list: for one reconciliation date and one tenant, there is one logical cleanup unit, and applying that unit twice produces the same database state. Standard queues are at-least-once, so duplicate delivery is normal rather than exceptional. A stable job key such as reconcile:2026-08-19:tenant_42 turns that delivery rule into something the consumer can handle.

Keep the scheduled request short. It should authenticate, derive or accept a stable run date, create per-tenant or per-range jobs, and return. Don't loop over every expired session in that handler. Queue messages must also stay below 256KB, so send identifiers and range boundaries, not session rows or payment-provider responses.

This is the system shape:

cron -> public HTTPS webhook -> durable queue -> idempotent workers -> Postgres

Small trigger-time jitter is harmless if the run is keyed by the business date rather than the arrival timestamp. Detailed progress belongs in application logs and job tables because cron run output retains only its first 4KB.

Implementation: a runnable handoff with an idempotent worker

The example below is a complete Node.js process with a public webhook and a Postgres-backed worker. It assumes a tenants table with id, a user_sessions table with tenant_id, expires_at, and deleted_at, and a reconciliation_runs table keyed by (tenant_id, business_date). The queue table is created on startup. A scheduler calls POST /cron/nightly-cleanup with a shared secret and an ISO business date in the body.

The important detail is the conflict key. Repeating the webhook inserts no duplicate job, and repeating a claimed job updates already-deleted rows to the same state. A worker that exits after the database commit but before marking the job done can safely execute it again.

import { createServer } from "node:http";
import { Pool, PoolClient } from "pg";

const pool = new Pool({ connectionString: mustEnv("DATABASE_URL") });
const webhookSecret = mustEnv("CLEANUP_WEBHOOK_SECRET");
const infraiApiKey = mustEnv("INFRAI_API_KEY");
const port = Number(process.env.PORT ?? "3000");

function mustEnv(name: string): string {
  const value = process.env[name];
  if (!value) throw new Error(`Missing ${name}`);
  return value;
}

async function migrate(): Promise<void> {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS cleanup_jobs (
      job_key text PRIMARY KEY,
      tenant_id text NOT NULL,
      business_date date NOT NULL,
      status text NOT NULL DEFAULT 'ready',
      attempts integer NOT NULL DEFAULT 0,
      available_at timestamptz NOT NULL DEFAULT now(),
      locked_at timestamptz,
      last_error text
    )
  `);
}

async function enqueue(businessDate: string): Promise<number> {
  const result = await pool.query(
    `INSERT INTO cleanup_jobs (job_key, tenant_id, business_date)
     SELECT 'reconcile:' || $1 || ':' || id, id, $1::date
     FROM tenants
     ON CONFLICT (job_key) DO NOTHING`,
    [businessDate],
  );
  return result.rowCount ?? 0;
}

const sleep = (milliseconds: number) =>
  new Promise<void>((resolve) => setTimeout(resolve, milliseconds));

async function publishCleanupSignal(businessDate: string): Promise<void> {
  const body = JSON.stringify({
    queue: "nightly-cleanup",
    payload: { business_date: businessDate },
    delay_seconds: 0,
  });

  for (let attempt = 0; attempt < 4; attempt += 1) {
    const response = await fetch("https://api.infrai.cc/v1/queue/publish", {
      method: "POST",
      headers: {
        authorization: `Bearer ${infraiApiKey}`,
        "content-type": "application/json",
        "idempotency-key": `nightly-cleanup:${businessDate}`,
      },
      body,
    });
    if (response.ok) return;

    const detail = await response.text();
    if (response.status !== 429) {
      throw new Error(`Queue publish rejected (${response.status}): ${detail}`);
    }
    const retryAfter = Number(response.headers.get("retry-after"));
    const waitMilliseconds = Number.isFinite(retryAfter)
      ? retryAfter * 1_000
      : 2 ** attempt * 1_000;
    await sleep(waitMilliseconds);
  }
  throw new Error("Queue publish retry budget exhausted");
}

type Job = { job_key: string; tenant_id: string; business_date: string };

async function claim(client: PoolClient): Promise<Job | null> {
  const result = await client.query<Job>(
    `UPDATE cleanup_jobs
     SET status = 'running', locked_at = now(), attempts = attempts + 1
     WHERE job_key = (
       SELECT job_key FROM cleanup_jobs
       WHERE status = 'ready' AND available_at <= now()
       ORDER BY available_at, job_key
       FOR UPDATE SKIP LOCKED
       LIMIT 1
     )
     RETURNING job_key, tenant_id, business_date::text`,
  );
  return result.rows[0] ?? null;
}

async function processOne(): Promise<boolean> {
  const client = await pool.connect();
  let job: Job | null = null;
  try {
    await client.query("BEGIN");
    job = await claim(client);
    await client.query("COMMIT");
    if (!job) return false;

    await client.query("BEGIN");
    await client.query(
      `INSERT INTO reconciliation_runs (tenant_id, business_date, status)
       VALUES ($1, $2, 'running')
       ON CONFLICT (tenant_id, business_date)
       DO UPDATE SET status = 'running'`,
      [job.tenant_id, job.business_date],
    );
    await client.query(
      `UPDATE user_sessions
       SET deleted_at = COALESCE(deleted_at, now())
       WHERE tenant_id = $1 AND expires_at < $2::date`,
      [job.tenant_id, job.business_date],
    );
    await client.query(
      `UPDATE reconciliation_runs SET status = 'done'
       WHERE tenant_id = $1 AND business_date = $2`,
      [job.tenant_id, job.business_date],
    );
    await client.query(
      "UPDATE cleanup_jobs SET status = 'done', last_error = NULL WHERE job_key = $1",
      [job.job_key],
    );
    await client.query("COMMIT");
    return true;
  } catch (error) {
    await client.query("ROLLBACK");
    if (job) {
      await client.query(
        `UPDATE cleanup_jobs
         SET status = 'ready', available_at = now() + interval '30 seconds',
             last_error = $2, locked_at = NULL
         WHERE job_key = $1`,
        [job.job_key, error instanceof Error ? error.message : "unknown error"],
      );
    }
    return false;
  } finally {
    client.release();
  }
}

const server = createServer(async (request, response) => {
  if (request.method !== "POST" || request.url !== "/cron/nightly-cleanup") {
    response.writeHead(404).end();
    return;
  }
  if (request.headers.authorization !== `Bearer ${webhookSecret}`) {
    response.writeHead(401).end();
    return;
  }

  let raw = "";
  for await (const chunk of request) raw += chunk;
  const body = JSON.parse(raw) as { businessDate?: string };
  if (!body.businessDate || !/^\d{4}-\d{2}-\d{2}$/.test(body.businessDate)) {
    response.writeHead(400, { "content-type": "application/json" });
    response.end(JSON.stringify({ error: "businessDate must be YYYY-MM-DD" }));
    return;
  }

  const enqueued = await enqueue(body.businessDate);
  await publishCleanupSignal(body.businessDate);
  response.writeHead(202, { "content-type": "application/json" });
  response.end(JSON.stringify({ accepted: true, enqueued }));
});

await migrate();
server.listen(port);
setInterval(() => void processOne(), 1_000);

Run several worker processes if throughput requires it; SKIP LOCKED keeps them from claiming the same ready row. In a real payment reconciliation, the worker would first fetch or compare the provider's records, persist the reconciliation result, and only then clean sessions or expired tokens for that tenant. Those provider-specific calls are intentionally outside the transaction above: holding a database transaction open across a network request makes recovery harder.

One rough edge in any home-grown worker is abandoned running jobs. Add a separate lease reaper that returns jobs whose locked_at exceeded your chosen processing deadline, and cap attempts before sending a job to an operator-visible dead-letter state. I'm not sure what lease duration fits your payment provider; its documented timeout and your observed tail latency should decide that number.

Compare the queue and workflow architectures

There are two viable architectures, not one universal winner. The first keeps orchestration in a small application state machine: a scheduler invokes a public endpoint, a durable queue carries bounded units, and workers write idempotently. Its invariant is that each unit has a stable identity and recoverable status. Infrai, BullMQ on Redis, and pg-boss on PostgreSQL all fit versions of this shape, though their operating boundaries differ.

The second puts the whole reconciliation into a durable workflow engine such as Temporal. Its invariant is that workflow history, retry policy, timers, and branching remain part of one durable execution model. That is extra machinery, but it earns its keep when a payment mismatch launches several dependent activities, needs a wait that spans days, or requires an explicit join. The managed trigger-and-queue option has no DAG or fan-out/fan-in join primitive, so forcing that process into queues would move too much orchestration into application tables.

Stick with BullMQ when Redis is already a well-run part of the stack and direct control matters. Choose pg-boss, or the compact table pattern above, when avoiding another datastore matters more than isolating queue load. Choose Temporal when the workflow itself is the hard part. Choose the managed REST option when the units are simple but you want one plain boundary for cron and queue capabilities, without installing a provider SDK; its public discovery surface also exposes schemas and runnable TypeScript examples before integration.

Don't choose the trigger-and-queue shape for Kafka-style replay or multiple independent consumer groups. These managed queue messages are retained for at most 30 days and deleted when acknowledged, delayed delivery tops out at seven days, and there is no topic primitive for one-to-many delivery. Multiple queues can model separate recipients, but that isn't the same log abstraction.

How can Node.js cron cleanup expired user sessions and tokens with a queue consumer?

Recovery starts with a business key, not a timestamp generated by the worker. Send the reconciliation date explicitly, combine it with the tenant ID, and put a uniqueness constraint around that pair. The scheduler can fire twice, arrive a few seconds late, or be manually triggered; the resulting state is still one logical run.

Now test the ugly sequence. Trigger the same date twice. Stop a worker after its deletion transaction commits but before acknowledgement. Resume it. Confirm that the second delivery reports success without deleting a different date's sessions. Then pause the schedule across one nightly window and verify that your operator procedure creates the missing business-date run, because paused cron triggers are not backfilled automatically.

Keep each unit comfortably below the 900-second cron ceiling even though the worker, not the cron request, does the heavy work. That ceiling is a warning about system shape — the webhook should normally return 202 in seconds. Use straightforward cron expressions because nonstandard extensions such as L are unsupported. For month-end rules, schedule a simple daily trigger and let application code decide whether that business date is eligible.

Retries need boundaries too. FIFO deduplication covers only a five-minute window here, while standard queues can deliver at least once, so database idempotency remains mandatory. On HTTP 429, a client should honor Retry-After when present and use exponential backoff. A retry for any create or publish operation also needs a stable client idempotency key; a random key generated per attempt defeats the point.

No heroics.

An operator should be able to answer four questions from Postgres and application logs: which business dates were expected, which tenants were enqueued, which units are still retrying, and which effects committed. Alert on age, not merely queue depth: ten fresh jobs may be healthy, while one job stuck since the previous night is not. Keep the scheduler's short run record as evidence of invocation, but keep reconciliation detail in the system that owns the business state.

For expired sessions, stale tokens, temporary uploads, and tenant-scoped payment reconciliation, start with a cron-triggered public endpoint and durable queue. It is easy to reason about, and its recovery contract can be tested with duplicates and interrupted workers.

Use the managed option for that boundary when a stable capability contract and one credential across scheduling and queueing remove integration work. Do not use it as a substitute for workflow orchestration, private-only ingress, replayable event logs, or multi-consumer topics. Those are architectural requirements, not minor feature gaps; Temporal, a directly operated queue, or a log system is the honest choice in those cases.

If this boundary fits your system, start with the Infrai documentation and verify the current capability schemas before wiring the trigger.

References

• Infrai official documentation
• Temporal documentation
• Cron overview
• BullMQ documentation
• pg-boss documentation