triggerdotdev--trigger.dev/apps/webapp/app/v3/accountsWebhookWorker.server.ts

import { Worker as RedisWorker } from "@trigger.dev/redis-worker";
import { z } from "zod";
import { env } from "~/env.server";
import { logger } from "~/services/logger.server";
import { singleton } from "~/utils/singleton";
import { ssoController } from "~/services/sso.server";
import {
  applyDirectorySyncEffects,
  unqueuedProvisioningError,
} from "~/services/directorySyncEffects.server";

// Dedicated worker for inbound account-management webhooks. The webhook
// proxy route verifies the signature via the plugin and enqueues the
// parsed event here; this worker calls back into the plugin to apply the
// DB writes. The plugin owns the vendor-specific logic; the webapp owns
// the queue runtime (this file), mirroring `commonWorker.server.ts`.
//
// Vendor-neutral by design: the catalog/job names and payload shape carry
// no provider identity.
const PayloadSchema = z
  .object({
    id: z.string(),
    event: z.string(),
    data: z.unknown(),
  })
  // Zod v3 treats a `z.unknown()` field as optional, so a payload missing
  // `data` entirely would otherwise validate. Require the key's presence
  // at runtime via refine — `.nonoptional()` is a v4-only API and isn't
  // available on the classic `z` import this repo pins.
  .refine((payload) => "data" in payload, {
    message: "data is required",
    path: ["data"],
  });

function initializeWorker() {
  const redisOptions = {
    keyPrefix: "accounts-webhook:worker:",
    host: env.COMMON_WORKER_REDIS_HOST,
    port: env.COMMON_WORKER_REDIS_PORT,
    username: env.COMMON_WORKER_REDIS_USERNAME,
    password: env.COMMON_WORKER_REDIS_PASSWORD,
    enableAutoPipelining: true,
    ...(env.COMMON_WORKER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
  };

  const worker = new RedisWorker({
    name: "accounts-webhook-worker",
    redisOptions,
    catalog: {
      "account.webhook.event": {
        schema: PayloadSchema,
        visibilityTimeoutMs: 30_000,
        retry: { maxAttempts: 5 },
      },
    },
    concurrency: {
      workers: 2,
      tasksPerWorker: 4,
      limit: 8,
    },
    pollIntervalMs: 1_000,
    immediatePollIntervalMs: 50,
    shutdownTimeoutMs: 30_000,
    jobs: {
      "account.webhook.event": async ({ payload }) => {
        // The plugin returns a Result; throw on error so the worker
        // retries (a resolved err would otherwise be silently dropped).
        // `z.unknown()` infers `data?: unknown`, but the contract's
        // SsoWebhookEvent requires `data: unknown` — restate the fields
        // explicitly so the optional-vs-required shapes line up.
        const result = await ssoController.processWebhookEvent({
          id: payload.id,
          event: payload.event,
          data: payload.data,
        });
        if (result.isErr()) {
          throw new Error(`account webhook processing failed: ${result.error}`);
        }
        // Directory-sync events return membership effects to apply against
        // public.* tables (the plugin never writes those). A throw here
        // bubbles to the worker for retry; effects are idempotent.
        const { unqueuedUserIds } = await applyDirectorySyncEffects(result.value.effects);
        if (unqueuedUserIds.length > 0) {
          throw unqueuedProvisioningError(unqueuedUserIds);
        }
      },
    },
  });

  // Only poll on worker-role instances (same gate as commonWorker) and
  // only when the feature is enabled (no plugin loaded otherwise).
  if (env.COMMON_WORKER_ENABLED === "true" && env.SSO_ENABLED) {
    logger.debug(`👨‍🏭 Starting accounts webhook worker at host ${env.COMMON_WORKER_REDIS_HOST}`);
    worker.start();
  }

  return worker;
}

export const accountsWebhookWorker = singleton("accountsWebhookWorker", initializeWorker);