---
title: "Queue Workers"
description: "Dub implements a resilient queue worker and event-streaming architecture built around Upstash Redis streams and QStash job dispatching. High-frequency telemetry—such as link clicks and workspace us..."
last_updated: "2026-10-12T07:00:10.139631+00:00"
canonical_url: "https://www.doc0.dev/docs/934e554a-e6a1-476f-bb2f-23e62d86c3fd/technical/background-processing/queue-workers"
---

<details>
<summary>Relevant source files</summary>

The following files were used as context for generating this wiki page:

- [apps/web/app/ee/api/cron/streams/send-link-clicked-webhooks/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/send-link-clicked-webhooks/route.ts)
- [apps/web/app/ee/api/cron/streams/update-click-stats/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-click-stats/route.ts)
- [apps/web/app/ee/api/cron/streams/update-workspace-links-usage/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-workspace-links-usage/route.ts)
- [apps/web/app/ee/api/cron/export/events/workspace/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/export/events/workspace/route.ts)
- [apps/web/app/ee/api/cron/usage/utils.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/usage/utils.ts)
- [apps/web/lib/upstash/redis-streams/workspace-links-usage.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/workspace-links-usage.ts)
- [apps/web/app/ee/api/cron/queue/retry/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/queue/retry/route.ts)
- [apps/web/app/ee/api/cron/rewards/process/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/rewards/process/route.ts)
- [apps/web/app/ee/api/cron/domains/transfer/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/domains/transfer/route.ts)
- [apps/web/lib/upstash/redis-streams/workspace-click-events.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/workspace-click-events.ts)
- [apps/web/lib/upstash/redis-streams/link-click-events.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/link-click-events.ts)
- [apps/web/lib/jobs/outbox.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts)
- [apps/web/app/ee/api/cloudflare/webhook/r2-object-created/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cloudflare/webhook/r2-object-created/route.ts)
- [apps/web/app/ee/api/cron/sync-redis-resources/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/sync-redis-resources/route.ts)
- [apps/web/app/ee/api/cron/aggregate-clicks/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/aggregate-clicks/route.ts)
- [apps/web/lib/jobs/send-workflows.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/send-workflows.ts)
- [apps/web/app/ee/api/stripe/webhook/utils/update-workspace-plan.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/stripe/webhook/utils/update-workspace-plan.ts)
- [apps/web/app/ee/api/cron/workspaces/delete/utils.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/workspaces/delete/utils.ts)
- [apps/web/app/ee/api/cron/streams/update-partner-stats/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-partner-stats/route.ts)
- [apps/web/app/ee/api/cron/cleanup/link-retention/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/cleanup/link-retention/route.ts)
- [apps/web/lib/jobs/handlers/process-shopify-order-job.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/handlers/process-shopify-order-job.ts)
- [apps/web/lib/webhook/failure.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/webhook/failure.ts)
- [apps/web/lib/webhook/click-webhook-workspaces.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/webhook/click-webhook-workspaces.ts)
- [apps/web/app/api/jobs/process/jobName/route.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/api/jobs/process/%5BjobName%5D/route.ts)
- [apps/web/lib/upstash/redis-streams/client.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/client.ts)
- [apps/web/lib/api/rewards/queue-reward-processing.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/api/rewards/queue-reward-processing.ts)
- [apps/web/app/ee/api/cron/sync-redis-resources/cleanup-redundant-link-webhook-entries.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/sync-redis-resources/cleanup-redundant-link-webhook-entries.ts)
- [apps/web/lib/jobs/handlers/partner-search-sync-job.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/handlers/partner-search-sync-job.ts)
- [apps/web/lib/integrations/shopify/checkout-cache.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/integrations/shopify/checkout-cache.ts)
- [apps/web/lib/jobs/handlers/process-partner-group-change-job.ts](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/handlers/process-partner-group-change-job.ts)
</details>

## Overview

Dub implements a resilient queue worker and event-streaming architecture built around Upstash Redis streams and QStash job dispatching. High-frequency telemetry—such as link clicks and workspace usage updates—is ingested into Redis streams and processed in batch consumers by cron routines with distributed locking. For transactional background tasks, the system uses a database-backed outbox pattern that persists failed QStash publishes for automated retry recovery. Asynchronous jobs route through dynamic endpoints and type-safe job definitions to execute domain-specific workflows for partner searches, order processing, and aggregate analytics.

Sources: [apps/web/app/ee/api/cron/streams/send-link-clicked-webhooks/route.ts:5-65](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/send-link-clicked-webhooks/route.ts#L5-L65), [apps/web/app/ee/api/cron/streams/update-click-stats/route.ts:3-112](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-click-stats/route.ts#L3-L112), [apps/web/app/ee/api/cron/queue/retry/route.ts:1-46](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/queue/retry/route.ts#L1-L46), [apps/web/lib/jobs/outbox.ts:88-122](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts#L88-L122), [apps/web/app/api/jobs/process/jobName/route.ts:11-105](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/api/jobs/process/%5BjobName%5D/route.ts#L11-L105)

## Redis Stream Pipeline Infrastructure

### Stream Operations and Client Abstraction

The `RedisStream` class encapsulates interaction with Upstash Redis streams, providing methods for batch message reading, transactional deletion, and metadata inspection. It initializes with a designated `streamKey` and exposes `processBatch` to fetch ranges of stream entries using `redis.xrange`, execute a handler function against them, and optionally delete successfully processed messages via `redis.xdel` to prevent memory accumulation.

```typescript
export class RedisStream {
  private streamKey: string;

  constructor(streamKey: string) {
    this.streamKey = streamKey;
  }

  /**
   * Read and process a batch of items from the stream
   * This provides a consumable batch of messages with the ability to delete them upon successful completion of the handler
   */
  async processBatch<T>(
    handler: (
      records: RedisStreamEntry<T>[],
    ) => Promise<any & { processedEntryIds: string[] }>,
    options: {
      startId?: string;
      endId?: string;
      count?: number;
      deleteAfterRead?: boolean;
    } = {},
  ): Promise<any> {
    const {
      startId = "0",
      endId = "+",
      count = 1000,
      deleteAfterRead = true,
    } = options;

    try {
      // Read entries from the stream
      const entriesMap = await redis.xrange(
        this.streamKey,
        startId,
        endId,
        count,
      );

      const entries: RedisStreamEntry<T>[] = Object.entries(entriesMap).map(
        ([id, data]) => ({
          id,
          data: data as any,
        }),
      );

      const { processedEntryIds, ...response } = await handler(entries);

      // Optionally delete processed entries to prevent memory buildup
      if (deleteAfterRead) {
        try {
          if (processedEntryIds.length > 0) {
            // xdel supports deleting multiple IDs at once
            await redis.xdel(this.streamKey, processedEntryIds);
          }
        } catch (error) {
          console.warn("Failed to clean up processed stream entries:", error);
          // Don't throw - this is not critical
        }
      }

      return response as Response;
    } catch (error) {
      console.error(
        "Failed to read workspace usage updates from stream:",
        error,
      );
      throw error;
    }
  }
```
Sources: [apps/web/lib/upstash/redis-streams/client.ts:8-76](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/client.ts#L8-L76)

The client also provides a `getStreamInfo` method to inspect the current state of a stream by querying its length, first entry ID via `redis.xrange`, and last entry ID via `redis.xrevrange`. If a stream does not exist or remains empty, the inspection catches the error state and returns zeroed metrics.

```typescript
/**
 * Get stream information (length, first/last entry, etc.)
 */
async getStreamInfo(): Promise<{
  length: number;
  firstEntryId: string | null;
  lastEntryId: string | null;
}> {
  try {
    // Get stream length - check if stream exists first
    let length = 0;
    let firstEntryId: string | null = null;
    let lastEntryId: string | null = null;

    try {
      // Try to get first entry to check if stream exists
      const firstEntry = await redis.xrange(this.streamKey, "-", "+", 1);

      if (firstEntry && Object.keys(firstEntry).length > 0) {
        firstEntryId = Object.keys(firstEntry)[0];

        // If stream exists, get its length and last entry
        const entries = await redis.xrange(this.streamKey, "-", "+");
        length = Object.keys(entries || {}).length;

        if (length > 0) {
          // Use "$" to get the last entry instead of "+"
          const lastEntry = await redis.xrevrange(
            this.streamKey,
            "+",
            "-",
            1,
          );

          if (lastEntry && Object.keys(lastEntry).length > 0) {
            const entryId = Object.keys(lastEntry)[0];

            if (entryId !== firstEntryId) {
              lastEntryId = entryId;
            }
          }
        }
      }
    } catch (streamError) {
      // Stream might not exist yet, which is fine
      console.log("Stream does not exist or is empty:", streamError);
      console.log(streamError.message);
      console.log(streamError.type);
      console.log(JSON.stringify(streamError, null, 2));
    }

    return {
      length,
      firstEntryId,
      lastEntryId,
    };
  } catch (error) {
    console.error("Failed to get stream info:", error);
    throw error;
  }
}
```
Sources: [apps/web/lib/upstash/redis-streams/client.ts:78-138](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/client.ts#L78-L138)

> [!NOTE]
> `RedisStream` handles missing streams gracefully during inspection by catching stream errors and returning null identifiers with a zero length, preventing worker crashes when polling uninitialized streams.

Sources: [apps/web/lib/upstash/redis-streams/client.ts:121-127](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/client.ts#L121-L127)

### Event Publishers and Database Fallbacks

Three primary stream publishers handle high-throughput event ingestion for link clicks, workspace click telemetry, and workspace link usage counts. Each publisher attempts an asynchronous `xadd` operation onto its respective Upstash Redis stream key, falling back to direct database execution if the Redis connection fails.

| Stream Key Constant | Module File | Payload Interface / Schema | Database Fallback Action |
| :--- | :--- | :--- | :--- |
| `link:click:events` | `link-click-events.ts` | `LinkClickEvent` | `UPDATE Link SET clicks = clicks + 1...` & optional Project/Program update |
| `workspace:click:events` | `workspace-click-events.ts` | `clickEventSchemaTB` | Log error to Axiom and rethrow |
| `workspace:linksUsage:updates` | `workspace-links-usage.ts` | `WorkspaceLinksUsageEvent` | `UPDATE Project SET linksUsage = linksUsage + ?, totalLinks = totalLinks + ? WHERE id = ?` |

Sources: [apps/web/lib/upstash/redis-streams/link-click-events.ts:6-73](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/link-click-events.ts#L6-L73), [apps/web/lib/upstash/redis-streams/workspace-click-events.ts:7-43](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/workspace-click-events.ts#L7-L43), [apps/web/lib/upstash/redis-streams/workspace-links-usage.ts:6-40](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/workspace-links-usage.ts#L6-L40)

The `publishLinkClickEvent` function constructs payloads containing `linkId`, `timestamp`, and optional `workspaceId`, `programId`, and `partnerId`. If publishing fails, it executes `Promise.allSettled` to increment link click counters and workspace usage directly in PlanetScale while flushing error logs.

```typescript
export const publishLinkClickEvent = async ({
  linkId,
  timestamp,
  workspaceId,
  programId,
  partnerId,
}: LinkClickEvent) => {
  const payload = {
    linkId,
    timestamp,
    ...(workspaceId && { workspaceId }),
    ...(programId && partnerId && { programId, partnerId }),
  };

  try {
    return await redis.xadd(STREAM_KEY, "*", payload);
  } catch (error) {
    logger.error("stream.publish_failed", {
      service: "upstash",
      streamKey: STREAM_KEY,
      error: toErrorFields(error),
      correlation: {
        linkId,
        workspaceId,
        programId,
        partnerId,
      },
    });

    return await Promise.allSettled([
      conn.execute(
        "UPDATE Link SET clicks = clicks + 1, lastClicked = NOW() WHERE id = ?",
        [linkId],
      ),

      workspaceId
        ? conn.execute(
            "UPDATE Project p SET p.usage = p.usage + 1, p.totalClicks = p.totalClicks + 1 WHERE id = ?",
            [workspaceId],
          )
        : null,

      programId && partnerId
        ? conn.execute(
            "UPDATE ProgramEnrollment SET totalClicks = totalClicks + 1 WHERE programId = ? AND partnerId = ?",
            [programId, partnerId],
          )
        : null,

      logger.flush(),
    ]);
  }
```
Sources: [apps/web/lib/upstash/redis-streams/link-click-events.ts:22-73](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/link-click-events.ts#L22-L73)

> [!WARNING]
> `publishWorkspaceClickEvent` validates incoming events against the `clickEventSchemaTB` Zod schema and checks `clickWebhookWorkspaces` membership before adding to the stream; events for workspaces without active `link.clicked` webhooks are dropped entirely.

Sources: [apps/web/lib/upstash/redis-streams/workspace-click-events.ts:12-26](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/upstash/redis-streams/workspace-click-events.ts#L12-L26)

## Link Click Event Processing

### Link Click Event Processing and Webhook Dispatches

Consuming click event streams involves cron-triggered batch workers that drain Redis streams, aggregate statistics into PlanetScale, and dispatch workspace webhook notifications via QStash. Two independent cron routines handle these operations: `update-click-stats` processes high-throughput click telemetry to increment link and workspace counters, while `send-link-clicked-webhooks` scans workspace click streams to trigger asynchronous delivery.

| Cron Route Path | Redis Stream Key | Lock Key | Lock TTL (s) | Batch Size |
| :--- | :--- | :--- | :--- | :--- |
| `/api/cron/streams/update-click-stats` | `link:click:events` | `lock:update-click-stats` | 600 | 10,000 |
| `/api/cron/streams/send-link-clicked-webhooks` | `workspace:click:events` | `lock:send-link-clicked-webhooks` | 600 | 6,000 |

Sources: [apps/web/app/ee/api/cron/streams/send-link-clicked-webhooks/route.ts:18-27](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/send-link-clicked-webhooks/route.ts#L18-L27), [apps/web/app/ee/api/cron/streams/update-click-stats/route.ts:15-20](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-click-stats/route.ts#L15-L20)

> [!CAUTION]
> Cron handlers acquire a distributed Redis lock with `nx: true` and an expiration matching the Vercel execution maximum duration (`600s`). If a run fails to acquire `lock:send-link-clicked-webhooks` or `lock:update-click-stats`, the execution exits immediately to prevent concurrent duplicate updates or double-counted webhook deliveries.

Sources: [apps/web/app/ee/api/cron/streams/send-link-clicked-webhooks/route.ts:32-42](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/send-link-clicked-webhooks/route.ts#L32-L42), [apps/web/app/ee/api/cron/streams/update-click-stats/route.ts:19-20](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-click-stats/route.ts#L19-L20)

### Aggregating Link and Partner Statistics

The `update-click-stats` cron job reads up to 10,000 entries per run from the `link:click:events` stream using `linkClickEventStream.processBatch`. The `aggregateClickStats` function groups incoming entries into map structures to batch database writes across three distinct aggregates: link click counters, workspace totals, and program enrollment stats.

```typescript
const aggregateClickStats = (
  entries: RedisStreamEntry<LinkClickEvent>[],
): {
  linkUpdates: LinkAggregate[];
  workspaceUpdates: WorkspaceAggregate[];
  enrollmentUpdates: EnrollmentAggregate[];
} => {
  const links = new Map<string, LinkAggregate>();
  const workspaces = new Map<string, WorkspaceAggregate>();
  const enrollments = new Map<string, EnrollmentAggregate>();

  for (const entry of entries) {
    const { linkId, workspaceId, programId, partnerId, timestamp } = entry.data;

    if (!linkId) {
      continue;
    }

    const parsedTimestamp = Date.parse(timestamp);
    const lastClicked = Number.isFinite(parsedTimestamp)
      ? parsedTimestamp
      : Date.now();

    const existingLink = links.get(linkId);
    if (existingLink) {
      existingLink.clicks += 1;
      existingLink.lastClicked = Math.max(
        existingLink.lastClicked,
        lastClicked,
      );
      existingLink.entryIds.push(entry.id);
    } else {
      links.set(linkId, {
        linkId,
        clicks: 1,
        lastClicked,
        entryIds: [entry.id],
      });
    }

    if (workspaceId) {
      const existingWorkspace = workspaces.get(workspaceId);
      if (existingWorkspace) {
        existingWorkspace.clicks += 1;
      } else {
        workspaces.set(workspaceId, {
          workspaceId,
          clicks: 1,
        });
      }
    }

    if (programId && partnerId) {
      const key = `${programId}:${partnerId}`;
      const existingEnrollment = enrollments.get(key);
      if (existingEnrollment) {
        existingEnrollment.clicks += 1;
      } else {
        enrollments.set(key, {
          programId,
          partnerId,
          clicks: 1,
        });
      }
    }
  }

  return {
    linkUpdates: Array.from(links.values()),
    workspaceUpdates: Array.from(workspaces.values()),
    enrollmentUpdates: Array.from(enrollments.values()),
  };
};
```
Sources: [apps/web/app/ee/api/cron/streams/update-click-stats/route.ts:40-112](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-click-stats/route.ts#L40-L112)

Database updates execute within parallel sub-batches of 50 items using `processInSubBatches`. Link click increments execute raw SQL statements against PlanetScale to update row totals and reconcile timestamps with `GREATEST`.

```typescript
const processInSubBatches = async <T>(
  items: T[],
  handler: (item: T) => Promise<{ success: boolean; error?: unknown }>,
) => {
  const errors: unknown[] = [];
  let totalProcessed = 0;

  for (let i = 0; i < items.length; i += SUB_BATCH_SIZE) {
    const batch = items.slice(i, i + SUB_BATCH_SIZE);
    const results = await Promise.allSettled(batch.map(handler));

    for (const result of results) {
      if (result.status === "fulfilled" && result.value.success) {
        totalProcessed++;
      } else if (result.status === "fulfilled" && result.value.error) {
        errors.push(result.value.error);
      } else if (result.status === "rejected") {
        errors.push(result.reason);
      }
    }
  }

  return { totalProcessed, errors };
};

const getStreamEntryAgeMs = (entryId: string | null) => {
  if (!entryId) {
    return null;
  }

  const timestampMs = Number(entryId.split("-")[0]);
  if (!Number.isFinite(timestampMs)) {
    return null;
  }

  return Date.now() - timestampMs;
};

const processClickStatsStreamBatch = () =>
  linkClickEventStream.processBatch<LinkClickEvent>(
    async (entries) => {
      if (!entries || entries.length === 0) {
        return {
          success: true,
          linkUpdates: [],
          workspaceUpdates: [],
          enrollmentUpdates: [],
          processedEntryIds: [],
          totalProcessed: 0,
          errors: [],
        };
      }

      console.log(`Aggregating ${entries.length} click stats events`);

      const { linkUpdates, workspaceUpdates, enrollmentUpdates } =
        aggregateClickStats(entries);

      if (linkUpdates.length === 0) {
        console.log("No click stats updates to process");
        return {
          success: true,
          linkUpdates: [],
          workspaceUpdates: [],
          enrollmentUpdates: [],
          processedEntryIds: entries.map((entry) => entry.id),
          totalProcessed: 0,
          errors: [],
        };
      }

      console.log(
        `Processing ${linkUpdates.length} link, ${workspaceUpdates.length} workspace, ${enrollmentUpdates.length} enrollment click stats updates...`,
      );

      const processedEntryIds: string[] = [];
      const errors: unknown[] = [];

      const linkResult = await processInSubBatches(
        linkUpdates,
```
Sources: [apps/web/app/ee/api/cron/streams/update-click-stats/route.ts:114-215](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-click-stats/route.ts#L114-L215)

> [!TIP]
> Partner activity statistics in `update-partner-stats` filter out undefined and non-finite numeric values before constructing dynamic `UPDATE ProgramEnrollment` queries, formatting JavaScript Date instances into `yyyy-MM-dd HH:mm:ss` strings.

Sources: [apps/web/app/ee/api/cron/streams/update-partner-stats/route.ts:277-301](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-partner-stats/route.ts#L277-L301)

### Workspace Webhook Dispatch Workflows

The `send-link-clicked-webhooks` worker drains the `workspace:click:events` stream, batching up to 6,000 entries per run. For each event batch, it queries Prisma for matching workspaces, links, and associated webhook configurations filtered by `LINK_CLICK_WEBHOOK_TRIGGER` and non-null `disabledAt` timestamps.

```typescript
const processStreamBatch = (): Promise<{
  processed: number;
  sent: number;
  failed: number;
  processedEntryIds: string[];
}> =>
  workspaceClickEventStream.processBatch<ClickEventTB>(
    async (entries) => {
      if (!entries || entries.length === 0) {
        return {
          processed: 0,
          sent: 0,
          failed: 0,
          processedEntryIds: [],
        };
      }

      const workspaceIds = [
        ...new Set(entries.map((e) => e.data.workspace_id).filter(Boolean)),
      ];
      const linkIds = [
        ...new Set(entries.map((e) => e.data.link_id).filter(Boolean)),
      ];

      const [workspaces, links] = await Promise.all([
        prisma.project.findMany({
          where: {
            id: {
              in: workspaceIds,
            },
          },
          select: {
            id: true,
            usage: true,
            usageLimit: true,
            webhooks: {
              where: {
                disabledAt: null,
                triggers: {
                  array_contains: [LINK_CLICK_WEBHOOK_TRIGGER],
                },
              },
              select: {
                id: true,
                url: true,
                secret: true,
                linkScope: true,
                projectId: true,
              },
            },
          },
        }),

        prisma.link.findMany({
          where: {
            id: {
              in: linkIds,
            },
          },
          include: {
            tags: {
              select: {
                tag: {
                  select: {
                    id: true,
                    name: true,
                    color: true,
                  },
                },
              },
            },
          },
        }),
      ]);

      const webhooks = workspaces.flatMap((w) => w.webhooks);
```
Sources: [apps/web/app/ee/api/cron/streams/send-link-clicked-webhooks/route.ts:67-142](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/send-link-clicked-webhooks/route.ts#L67-L142)

Webhooks support scoping rules via the `linkScope` property, segregating dispatches between link-specific associations (`linkWebhook`) and folder-level groupings (`folderWebhook`). Redundant entries are cleaned up periodically by `cleanupRedundantLinkWebhookEntries`, which removes orphaned `LinkWebhook` rows linked to non-link-scoped webhooks.

Sources: [apps/web/app/ee/api/cron/sync-redis-resources/cleanup-redundant-link-webhook-entries.ts:7-20](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/sync-redis-resources/cleanup-redundant-link-webhook-entries.ts#L7-L20)

## Workspace Link Usage Accounting

### Stream Consumer Workflows and Batching

The `update-workspace-links-usage` cron job consumes from the `workspace:links-usage` stream, pulling batches of up to 10,000 entries using `workspaceLinksUsageStream.processBatch`. The `aggregateWorkspaceLinksUsage` helper iterates over the raw stream entries, validating that each entry contains a valid `workspaceId` and a finite, positive `linksCount`. Valid items are grouped into a `Map` keyed by `workspaceId` to accumulate link counts and track associated stream entry IDs for acknowledgment.

```typescript
const aggregateWorkspaceLinksUsage = (
  entries: RedisStreamEntry<WorkspaceLinksUsageEvent>[],
): {
  updates: WorkspaceLinksUsageAggregate[];
  lastProcessedId: string | null;
} => {
  const aggregatedUsage = new Map<string, WorkspaceLinksUsageAggregate>();
  let lastProcessedId: string | null = null;

  for (const entry of entries) {
    const workspaceId = entry.data.workspaceId;
    const linksCount = Number(entry.data.linksCount);

    if (!workspaceId || !Number.isFinite(linksCount) || linksCount <= 0) {
      continue;
    }

    lastProcessedId = entry.id;

    if (aggregatedUsage.has(workspaceId)) {
      const existing = aggregatedUsage.get(workspaceId)!;
      existing.linksCount += linksCount;
      existing.entryIds.push(entry.id);
      continue;
    }

    aggregatedUsage.set(workspaceId, {
      workspaceId,
      linksCount,
      entryIds: [entry.id],
    });
  }

  return {
    updates: Array.from(aggregatedUsage.values()),
    lastProcessedId,
  };
};
```
Sources: [apps/web/app/ee/api/cron/streams/update-workspace-links-usage/route.ts:40-77](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-workspace-links-usage/route.ts#L40-L77)

The aggregated updates are split into sub-batches of 50 items and processed concurrently via `Promise.allSettled`. For each workspace in a sub-batch, Prisma updates the `Project` model by atomically incrementing both `linksUsage` and `totalLinks`, returning the updated usage and quota limits.

```typescript
const SUB_BATCH_SIZE = 50;
const batches: WorkspaceLinksUsageAggregate[][] = [];

for (let i = 0; i < updates.length; i += SUB_BATCH_SIZE) {
  batches.push(updates.slice(i, i + SUB_BATCH_SIZE));
}

const processedEntryIds: string[] = [];
const errors: { workspaceId: string; error: unknown }[] = [];
const updatedWorkspaces: (WorkspaceProps & {
  linksUsage: number;
  linksLimit: number;
})[] = [];
let totalProcessed = 0;

for (const batch of batches) {
  const batchResults = await Promise.allSettled(
    batch.map(async (update) => {
      const workspace = await prisma.project.update({
        where: {
          id: update.workspaceId,
        },
        data: {
          linksUsage: {
            increment: update.linksCount,
          },
          totalLinks: {
            increment: update.linksCount,
          },
        },
        select: {
          id: true,
          name: true,
          slug: true,
          linksUsage: true,
          linksLimit: true,
          plan: true,
        },
      });

      processedEntryIds.push(...update.entryIds);
      return workspace;
    }),
  );
```
Sources: [apps/web/app/ee/api/cron/streams/update-workspace-links-usage/route.ts:101-144](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-workspace-links-usage/route.ts#L101-L144)

> [!WARNING]
> If a workspace update fails within `Promise.allSettled`, the failure is captured in an `errors` array alongside the `workspaceId`, but execution continues for the remainder of the sub-batch. Successfully processed entry IDs are accumulated independently to ensure downstream acknowledgment succeeds for valid writes.

Sources: [apps/web/app/ee/api/cron/streams/update-workspace-links-usage/route.ts:146-160](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-workspace-links-usage/route.ts#L146-L160)

### Quota Thresholds and Limit Alerts

After updating workspace link counters, the worker calculates the percentage of quota consumed against each workspace's `linksLimit`. The `getLimitEmailType` utility function maps consumption percentages to notification types: percentages below 80% return `null`, values from 80% up to 100% return `firstLinksLimitEmail`, and values reaching or exceeding 100% return `secondLinksLimitEmail`.

| Percentage Range | Mapped Limit Email Type |
| :--- | :--- |
| `< 80%` | `null` (No alert) |
| `80% ≤ percentage < 100%` | `firstLinksLimitEmail` |
| `≥ 100%` | `secondLinksLimitEmail` |

Sources: [apps/web/app/ee/api/cron/streams/update-workspace-links-usage/route.ts:28-38](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-workspace-links-usage/route.ts#L28-L38), [apps/web/app/ee/api/cron/streams/update-workspace-links-usage/route.ts:162-184](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-workspace-links-usage/route.ts#L162-L184)

Workspaces crossing the alert threshold are filtered against previously dispatched notifications stored in the `sentEmail` table. Simultaneously, team member emails and workspace-level Slack webhooks are fetched in parallel to dispatch limit warnings.

```typescript
const workspaceIds = workspacesAboveThreshold.map(
  ({ workspace }) => workspace.id,
);
let notificationsSent = 0;

if (workspaceIds.length > 0) {
  const [sentEmails, users, slackWebhookByWorkspace] = await Promise.all([
    prisma.sentEmail.findMany({
      where: {
        projectId: {
          in: workspaceIds,
        },
        type: {
          in: ["firstLinksLimitEmail", "secondLinksLimitEmail"],
        },
      },
      select: {
        projectId: true,
        type: true,
      },
    }),

    prisma.user.findMany({
      where: {
        projects: {
          some: {
            projectId: {
              in: workspaceIds,
            },
          },
        },
      },
      select: {
        email: true,
        projects: {
          where: {
            projectId: {
              in: workspaceIds,
            },
          },
          select: {
            projectId: true,
          },
        },
      },
    }),

    getSlackWebhooks(workspaceIds),
  ]);

```
Sources: [apps/web/app/ee/api/cron/streams/update-workspace-links-usage/route.ts:185-234](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/streams/update-workspace-links-usage/route.ts#L185-L234)

## Transactional Job Outbox Pattern

### Job Persistence and Deferred Execution

The outbox pattern ensures guaranteed delivery of background jobs and workflows by persisting failed dispatches to the Prisma `Job` model before retry attempts. When background jobs are created via `persistBackgroundJobs`, `scheduledAt` timestamps are computed from options such as `notBefore` or `delay` offsets, and records are written with unique identifiers generated with the `job_` prefix.

```typescript
export async function persistBackgroundJobs(inputs: DispatchJobInput[]) {
  const jobs: PersistableJob[] = inputs.map(({ name, payload, options }) => {
    let scheduledAt = new Date();

    if (options?.notBefore) {
      scheduledAt = new Date(options.notBefore * 1000);
    } else if (typeof options?.delay === "number") {
      scheduledAt = new Date(Date.now() + options.delay * 1000);
    }

    const replayOptions = {
      ...(options?.deduplicationId && {
        deduplicationId: options.deduplicationId,
      }),
      ...(options?.retries !== undefined && { retries: options.retries }),
      ...(options?.queue && { queue: options.queue }),
      ...(options?.flowControl && { flowControl: options.flowControl }),
      ...(options?.label && { label: options.label }),
    };

    return {
      id: createId({ prefix: "job_" }),
      name,
      payload,
      options: replayOptions,
      scheduledAt,
    };
  });

  await prisma.job.createMany({
    data: jobs.map((job) => toJobCreateInput({ job })),
  });

  return jobs;
}
```
Sources: [apps/web/lib/jobs/outbox.ts:88-122](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts#L88-L122)

Two primary transports handle job distribution through `transports`: workflow triggers mapped via `isWorkflowName` and standard job dispatches matched via `isDefineJobName`. The `publishPendingJobs` function queries due rows where `scheduledAt` is less than or equal to the current date and `attempts` is strictly less than `MAX_JOB_ATTEMPTS`, ordered by creation date ascending up to `MAX_JOBS_PER_BATCH`.

```typescript
type JobTransport = {
  matches: (name: string) => boolean;
  send: (jobs: Job[]) => Promise<PublishResult[]>;
};

const transports: JobTransport[] = [
  {
    matches: isWorkflowName,
    send: triggerWorkflows,
  },
  {
    matches: isDefineJobName,
    send: sendJobs,
  },
];
```
Sources: [apps/web/lib/jobs/outbox.ts:10-24](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts#L10-L24), [apps/web/lib/jobs/outbox.ts:202-220](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts:202-L220)

> [!NOTE]
> The first failed publish attempt counts as attempt 1 so that the retry budget accurately matches `MAX_JOB_ATTEMPTS`. If a job name does not match any registered transport, it is treated as an unknown kind, generating a failed result with an error message and logging the event.

Sources: [apps/web/lib/jobs/outbox.ts:26-43](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts:26-43), [apps/web/lib/jobs/outbox.ts:245-261](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts:245-L261)

### Cron Retry Execution and Settle Mechanics

The cron endpoint `GET /api/cron/queue/retry` orchestrates periodic reprocessing of pending outbox items. A distributed Redis lock using `LOCK_KEY` (`lock:queue-retry`) with a TTL of 600 seconds prevents concurrent minute-cron overlapping invocations.

```typescript
const LOCK_KEY = "lock:queue-retry";
const LOCK_TTL_SECONDS = 600;

// GET /api/cron/queue/retry – republish jobs that failed to publish to QStash
// at dispatch time; rows are deleted on successful publish
export const GET = withCron(async () => {
  const acquired = await redis.set(LOCK_KEY, "1", {
    nx: true,
    ex: LOCK_TTL_SECONDS,
  });

  if (!acquired) {
    return logAndRespond(
      "[queue-retry] Another run is in progress. Skipping...",
    );
  }
```
Sources: [apps/web/app/ee/api/cron/queue/retry/route.ts:11-26](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/queue/retry/route.ts#L11-L26)

Once the lock is acquired, `publishPendingJobs` runs and delegates results to `settlePublishResults`. Successfully published job IDs are deleted from the database via `prisma.job.deleteMany`, while failed dispatches increment their attempt counters and update `lastError` fields.

```typescript
try {
  const { attempted, published, failed } = await publishPendingJobs();

  if (attempted === 0) {
    return logAndRespond("No background jobs to retry.");
  }

  if (failed === 0) {
    return logAndRespond(
      `Republished ${published} background jobs to QStash.`,
    );
  }

  return logAndRespond(
    `Republished ${published} background jobs to QStash; failed to republish ${failed}.`,
  );
} finally {
  await redis.del(LOCK_KEY);
}
```
Sources: [apps/web/app/ee/api/cron/queue/retry/route.ts:28-46](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/(ee)/api/cron/queue/retry/route.ts#L28-L46), [apps/web/lib/jobs/outbox.ts:125-180](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts:125-L180)

| Result Status | Action Taken | Prisma Operation |
| :--- | :--- | :--- |
| `published` | Delete successfully dispatched jobs | `prisma.job.deleteMany` |
| `failed` | Increment attempt counter and record `lastError` | `prisma.job.updateMany` |

Sources: [apps/web/lib/jobs/outbox.ts:125-180](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts:125-180)

> [!WARNING]
> When jobs exceed or equal `MAX_JOB_ATTEMPTS` during settlement failures, an error event (`jobs.retry_exhausted`) is logged to Axiom with the exhausted job identifiers and names, signaling permanent delivery abandonment.

Sources: [apps/web/lib/jobs/outbox.ts:182-194](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/outbox.ts:182-194)

## Asynchronous Job Handlers Execution

### Asynchronous Job Execution Router

The POST endpoint `POST /api/jobs/process/[jobName]` handles incoming background job execution requests originating from QStash triggers, with a maximum execution duration configured to 600 seconds. Every incoming request must pass cryptographic verification via `verifyQstashSignature` using the raw request body and request headers before any parsing or processing occurs. If signature verification fails, an error response is returned immediately.

```typescript
export const maxDuration = 600;

// POST /api/jobs/process/[jobName] – shared executor for all background jobs
export const POST = withAxiomBodyLog(
  async (
    req: Request,
    { params }: { params: Promise<{ jobName: string }> },
  ) => {
    const { jobName: urlJobName } = await params;
    const clonedReq = req.clone();
    const rawBody = await clonedReq.text();

    try {
      await verifyQstashSignature({ req, rawBody });
    } catch (error) {
      return handleAndReturnErrorResponse(error);
    }

```
Sources: [apps/web/app/api/jobs/process/jobName/route.ts:8-25](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/api/jobs/process/%5BjobName%5D/route.ts#L8-L25)

After successful verification, the request body is parsed as JSON and validated against the `jobEnvelopeSchema` envelope structure. If the envelope is malformed or invalid, an error is logged to Axiom and the endpoint returns a non-retryable response status to prevent QStash from retrying permanently corrupted payloads. The URL parameter `urlJobName` is also validated against the job name contained within the envelope data; any mismatch triggers an immediate non-retryable response.

```typescript
const envelope = jobEnvelopeSchema.safeParse(parsedBody);

// Permanently invalid — returning 2xx stops QStash retries
if (!envelope.success) {
  logger.error("jobs.invalid_envelope", {
    error: envelope.error.message,
  });
  await logger.flush();

  return new Response("Invalid job envelope (non-retryable).");
}

const { name: jobName } = envelope.data;

if (urlJobName !== jobName) {
  logger.error("jobs.job_name_mismatch", {
    urlJobName,
    envelopeJobName: jobName,
  });
  await logger.flush();

  return new Response(
    `Job name mismatch: URL "${urlJobName}" does not match envelope "${jobName}" (non-retryable).`,
  );
}
```
Sources: [apps/web/app/api/jobs/process/jobName/route.ts:34-58](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/api/jobs/process/%5BjobName%5D/route.ts#L34-L58)

Once the job name and envelope are verified, the executor retrieves the registered job definition using `loadJob(jobName)`. If no matching job definition is found in the registry, an unknown job error is logged and a non-retryable response is returned.

```typescript
console.log(`[jobs:${jobName}] Executing job...`, envelope.data);

// Get the job definition
const job = await loadJob(jobName);

if (!job) {
  logger.error("jobs.unknown_job", {
    jobName,
  });
  await logger.flush();

  return new Response(`Unknown job "${jobName}" (non-retryable).`);
}
```
Sources: [apps/web/app/api/jobs/process/jobName/route.ts:60-72](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/api/jobs/process/%5BjobName%5D/route.ts#L60-L72)

> [!WARNING]
> If a job's execution throws a Zod validation error (`z.ZodError`), the error is logged as an invalid payload and treated as permanent, returning a 2xx response so that QStash halts retries. Transient execution failures throw a 500 status code, signaling QStash to re-attempt delivery according to the job's retry policy.

Sources: [apps/web/app/api/jobs/process/jobName/route.ts:74-102](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/app/api/jobs/process/%5BjobName%5D/route.ts#L74-L102)

### Domain-Specific Job Handlers

Specific background job processors handle complex partner, group, and integration workflows dispatched through the central router. The `partnerSearchSyncJob` handles synchronization tasks with a parallelism limit of 20 concurrent executions using a flow control key (`partner-search-sync`), supporting a discriminated union input schema for either individual `enrollments` or bulk `partners` sync batches.

```typescript
const SYNC_PARALLELISM = 20;

/**
 * Two shapes, because the two kinds of change have different blast radii.
 *
 * `enrollments` carries IDs the caller already has, and is the only shape that
 * can express a deletion, since the handler finds out by failing to read one
 * back. `partners` is for changes that fan out past one enrollment, where the
 * caller knows the partner but not how many programs it reaches.
 */
const inputSchema = z.discriminatedUnion("type", [
  z.object({
    type: z.literal("enrollments"),
    enrollmentIds: z
      .array(z.string())
      .min(1)
      .max(PARTNER_SEARCH_SYNC_BATCH_SIZE),
  }),
  z.object({
    type: z.literal("partners"),
    partnerIds: z.array(z.string()).min(1).max(PARTNER_SEARCH_SYNC_BATCH_SIZE),
    programId: z.string().optional(),
    after: z.string().optional(),
  }),
]);

export const partnerSearchSyncJob = defineJob({
  name: "partner-search-sync-job",
  schema: inputSchema,
  defaults: {
    retries: 3,
    flowControl: {
      key: "partner-search-sync",
      parallelism: SYNC_PARALLELISM,
    },
  },
```
Sources: [apps/web/lib/jobs/handlers/partner-search-sync-job.ts:16-51](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/handlers/partner-search-sync-job.ts#L16-L51)

If no search provider is configured via `getPartnerSearchProvider()`, the search sync job exits early to prevent accumulating an unprocessable backlog. When processing `partners` input types, if the number of upserted items reaches the batch size limit `PARTNER_SEARCH_SYNC_BATCH_SIZE` alongside a valid `lastEnrollmentId`, the handler recursively dispatches another job instance with a 1-second delay.

```typescript
async handle(input) {
  const searchProvider = getPartnerSearchProvider();

  // An environment without the key should not accumulate a backlog of jobs
  // that can never succeed.
  if (!searchProvider) {
    console.log(
      "[partnerSearchSyncJob] No search provider configured. Skipping...",
    );
    return;
  }

  if (input.type === "enrollments") {
    const { upserted, deleted } = await syncPartnerSearchDocuments({
      enrollmentIds: input.enrollmentIds,
      searchProvider,
    });

    console.log(
      `[partnerSearchSyncJob] Synced ${upserted} and removed ${deleted} enrollment documents.`,
    );

    return;
  }

  const { upserted, lastEnrollmentId } = await syncPartnerEnrollments({
    partnerIds: input.partnerIds,
    programId: input.programId,
    after: input.after,
    take: PARTNER_SEARCH_SYNC_BATCH_SIZE,
    searchProvider,
  });

  if (upserted === 0) {
    return;
  }

  console.log(
    `[partnerSearchSyncJob] Synced ${upserted} enrollment documents for ${input.partnerIds.length} partner(s).`,
  );

  if (upserted === PARTNER_SEARCH_SYNC_BATCH_SIZE && lastEnrollmentId) {
    await partnerSearchSyncJob.dispatch(
      {
        ...input,
        after: lastEnrollmentId,
      },
      {
        delay: 1,
      },
    );
  }
},
```
Sources: [apps/web/lib/jobs/handlers/partner-search-sync-job.ts:52-104](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/handlers/partner-search-sync-job.ts#L52-L104)

```mermaid
flowchart TD
    Start["Job Execution Triggered"] --> CheckProvider{"Search Provider Configured?"}
    CheckProvider -- No --> ExitEarly["Log and Return Early"]
    CheckProvider -- Yes --> CheckType{"Input Type?"}
    CheckType -- "enrollments" --> SyncEnrollments["Sync Search Documents (upserted/deleted)"]
    CheckType -- "partners" --> SyncPartners["Sync Partner Enrollments"]
    SyncPartners --> CheckUpsert{"Upserted == Batch Size & lastEnrollmentId?"}
    CheckUpsert -- Yes --> DispatchNext["Dispatch Follow-up Job (delay: 1s)"]
    CheckUpsert -- No --> End["Complete"]
    SyncEnrollments --> End
```
Sources: [apps/web/lib/jobs/handlers/partner-search-sync-job.ts:52-104](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/handlers/partner-search-sync-job.ts#L52-L104)

Order and partner group modification workflows manage transactional updates and external integrations. The `processShopifyOrderJob` validates order schemas and workspace settings, capturing webhook logs with status codes 200 on success or 400 on error. Concurrently, `processPartnerGroupChangeJob` validates group existence, clears stale link rewards and Redis link caches, queues search index updates, remaps default links, records link analytics, and triggers partner notifications.

```typescript
export const processShopifyOrderJob = defineJob({
  name: "process-shopify-order-job",
  schema: inputSchema,
  async handle({ workspaceId, clickId, order }) {
    const startTime = Date.now();

    const workspace = await prisma.project.findUniqueOrThrow({
      where: {
        id: workspaceId,
      },
      select: {
        id: true,
        defaultProgramId: true,
        webhookEnabled: true,
      },
    });

    const requestLog = {
      workspaceId: workspace.id,
      method: "POST",
      path: "/shopify/integration/webhook" as const,
      requestBody: order,
      userAgent: "Shopify Order Job",
    };

    try {
      const result = await processShopifyOrder({
        order,
        workspace,
        clickId,
      });

      await logger.flush();

      await captureWebhookLog({
        ...requestLog,
        statusCode: 200,
        duration: Date.now() - startTime,
        responseBody: result,
      });
    } catch (error) {
      await captureWebhookLog({
        ...requestLog,
        statusCode: 400,
        duration: Date.now() - startTime,
        responseBody: serializeError(error),
      });

      await logger.flush();
    }
  },
```
Sources: [apps/web/lib/jobs/handlers/process-shopify-order-job.ts:17-67](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/handlers/process-shopify-order-job.ts#L17-L67), [apps/web/lib/jobs/handlers/process-partner-group-change-job.ts:28-145](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/jobs/handlers/process-partner-group-change-job.ts:28-L145)

## Webhook Retries and Failure Handling

### Consecutive Failure Tracking and Circuit Breaking

When a webhook delivery fails, `handleWebhookFailure` increments the `consecutiveFailures` counter and updates `lastFailedAt` on the targeted webhook record in the database. If the webhook has already been marked with a `disabledAt` timestamp, the function returns immediately without further processing.

```typescript
export const handleWebhookFailure = async (webhookId: string) => {
  const webhook = await prisma.webhook.update({
    where: {
      id: webhookId,
    },
    data: {
      consecutiveFailures: { increment: 1 },
      lastFailedAt: new Date(),
    },
    select: {
      id: true,
      url: true,
      secret: true,
      triggers: true,
      disabledAt: true,
      consecutiveFailures: true,
      lastFailedAt: true,
      projectId: true,
    },
  });

  if (webhook.disabledAt) {
    return;
  }
```
Sources: [apps/web/lib/webhook/failure.ts:12-35](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/webhook/failure.ts#L12-L35)

> [!NOTE]
> Successful deliveries reset the consecutive failure count and clear `lastFailedAt` back to `null` by calling `resetWebhookFailureCount`.

Sources: [apps/web/lib/webhook/failure.ts:65-73](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/webhook/failure.ts#L65-L73)

### Failure Notifications and Workspace Deactivation

As failures accumulate, the system dispatches notification emails to workspace owners and eventually deactivates the webhook once it breaches the disable threshold. If `consecutiveFailures` matches any value defined in `WEBHOOK_FAILURE_NOTIFY_THRESHOLDS`, `notifyWebhookFailure` queries the project's owner and sends a `WebhookFailed` email containing the webhook ID, URL, consecutive failure count, and disable threshold.

```typescript
if (
  WEBHOOK_FAILURE_NOTIFY_THRESHOLDS.includes(
    webhook.consecutiveFailures as any,
  )
) {
  await notifyWebhookFailure(webhook);
  return;
}
```
Sources: [apps/web/lib/webhook/failure.ts:37-44](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/webhook/failure.ts#L37-L44)

When `consecutiveFailures` reaches or exceeds `WEBHOOK_FAILURE_DISABLE_THRESHOLD`, the webhook is permanently disabled by setting its `disabledAt` field to the current timestamp. Concurrently, `Promise.allSettled` triggers `notifyWebhookDisabled` to email workspace owners via the `WebhookDisabled` template and invokes `syncWorkspaceWebhookStatus` to update the project's webhook-enabled status.

```typescript
  if (webhook.consecutiveFailures >= WEBHOOK_FAILURE_DISABLE_THRESHOLD) {
    // Disable the webhook
    const updatedWebhook = await prisma.webhook.update({
      where: { id: webhookId },
      data: {
        disabledAt: new Date(),
      },
    });

    await Promise.allSettled([
      // Notify the user
      notifyWebhookDisabled(updatedWebhook),

      // Update the project webhookEnabled flag
      syncWorkspaceWebhookStatus(webhook.projectId),
    ]);
  }
};
```
Sources: [apps/web/lib/webhook/failure.ts:46-63](https://github.com/blade47/dub/blob/7c93d97abf0ecd37fb669c6a68a1f7ef9109daea/apps/web/lib/webhook/failure.ts#L46-L63)

## Related

- [Scheduled Cron Tasks](https://www.doc0.dev/docs/934e554a-e6a1-476f-bb2f-23e62d86c3fd/technical/background-processing/scheduled-cron-tasks)
- [Webhooks & Postbacks](https://www.doc0.dev/docs/934e554a-e6a1-476f-bb2f-23e62d86c3fd/technical/background-processing/webhooks-postbacks)


## Sitemap

See the full [sitemap](https://www.doc0.dev/docs/934e554a-e6a1-476f-bb2f-23e62d86c3fd/llms.txt) for all pages in this wiki.