Queue Workers

Copyexport 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;
    }
  }

Copy/**
 * 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;
  }
}

Copyexport 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(),
    ]);
  }

Copyconst 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()),
  };
};

Copyconst 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,

Copyconst 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);

Copyconst 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,
  };
};

Copyconst 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;
    }),
  );

Copyconst 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),
  ]);
 

Copyexport 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;
}

Copytype JobTransport = {
  matches: (name: string) => boolean;
  send: (jobs: Job[]) => Promise<PublishResult[]>;
};
 
const transports: JobTransport[] = [
  {
    matches: isWorkflowName,
    send: triggerWorkflows,
  },
  {
    matches: isDefineJobName,
    send: sendJobs,
  },
];

Copyconst 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...",
    );
  }

Copytry {
  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);
}

Copyexport 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);
    }
 

Copyconst 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).`,
  );
}

Copyconsole.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).`);
}

Copyconst 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,
    },
  },

Copyasync 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,
      },
    );
  }
},

Copyexport 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();
    }
  },

Copyexport 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;
  }

Copyif (
  WEBHOOK_FAILURE_NOTIFY_THRESHOLDS.includes(
    webhook.consecutiveFailures as any,
  )
) {
  await notifyWebhookFailure(webhook);
  return;
}

Copy  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),
    ]);
  }
};