Ad Conversion Sync

Copyconst googleAdsApi = new GoogleAdsApi({
  accessToken: token.access_token,
});
 
const customers = await googleAdsApi.listAccessibleCustomers();
 
let customerName: string | null = null;
let customerId: string | null = null;
let loginCustomerId: string | null = null;
 
// Just one customer, so we can use the first one
if (customers.length === 1) {
  customerName = customers[0].descriptiveName;
  customerId = customers[0].id;
  loginCustomerId = inferLoginCustomerId({
    customers,
    selectedCustomerId: customerId,
  });
}

Copyexport const inferLoginCustomerId = ({
  customers,
  selectedCustomerId,
}: {
  customers: {
    id: string;
    manager: boolean;
    loginCustomerId?: string | null;
  }[];
  selectedCustomerId: string;
}) => {
  const normalizedSelectedId = selectedCustomerId.replace(/-/g, "");
  const selectedCustomer = customers.find(
    (customer) => customer.id.replace(/-/g, "") === normalizedSelectedId,
  );
 
  if (!selectedCustomer) {
    return null;
  }
 
  if (selectedCustomer.loginCustomerId) {
    return selectedCustomer.loginCustomerId.replace(/-/g, "");
  }
 
  if (selectedCustomer.manager) {
    return normalizedSelectedId;
  }
 
  return null;
};

Copyfor (const loginCustomerId of candidates) {
  try {
    const googleAdsApi = new GoogleAdsApi({
      accessToken: token.access_token,
      loginCustomerId,
      customerId,
    });
 
    const conversionActions =
      await googleAdsApi.listUploadClickConversionActions(customerId);
 
    return NextResponse.json({
      conversionActions,
      loginCustomerId,
    });
  } catch (error) {
    lastError = error;
 
    const message = error instanceof Error ? error.message : String(error);
 
    if (!message.includes("USER_PERMISSION_DENIED")) {
      throw error;
    }
  }
}

Copyconst extractGoogleAdsClickId = (url: string): GoogleAdsClickId | null => {
  try {
    const queryParams = getSearchParams(url);
 
    if (queryParams.gclid) {
      return {
        gclid: queryParams.gclid,
      };
    }
 
    if (queryParams.gbraid) {
      return {
        gbraid: queryParams.gbraid,
      };
    }
 
    if (queryParams.wbraid) {
      return {
        wbraid: queryParams.wbraid,
      };
    }
 
    return null;
  } catch {
    return null;
  }
};

Copyexport const queueGoogleAdsConversionUpload = async (
  payload: z.infer<typeof googleAdsConversionUploadSchema>,
) => {
  if (!extractGoogleAdsClickId(payload.click.url)) {
    return;
  }
 
  if (!(await googleAdsInstalledWorkspaces.has(payload.workspaceId))) {
    return;
  }
 
  // Data Manager expects major currency units, so we need to divide by 100
  // if the currency is NOT a zero decimal currency (e.g. JPY doesn't have cents)
  if (
    payload.conversionValue &&
    payload.currencyCode &&
    !isZeroDecimalCurrency(payload.currencyCode)
  ) {
    payload.conversionValue = payload.conversionValue / 100;
  }
 
  try {
    const response = await qstash.publishJSON({
      url: `${APP_DOMAIN_WITH_NGROK}/api/google-ads/upload-conversion`,
      body: payload,
      retries: 3,
      deduplicationId: `google-ads-${payload.workspaceId}-${payload.eventId}`,
    });
 
    if (!response.messageId) {
      throw new Error("Failed to queue Google Ads conversion upload");
    }
 
    return response;
  } catch (error) {
    logger.error("google-ads.queue_conversion_failed", {
      service: "google-ads",
      error: toErrorFields(error),
      correlation: {
        workspaceId: payload.workspaceId,
        eventId: payload.eventId,
        eventType: payload.eventType,
        clickId: payload.click.id,
      },
    });
 
    await logger.flush();
    throw error;
  }
};

Copyexport const uploadGoogleAdsConversion = async (
  payload: z.infer<typeof googleAdsConversionUploadSchema>,
): Promise<GoogleAdsConversionUploadResult> => {
  const {
    workspaceId,
    eventType,
    click,
    conversionDateTime,
    eventId,
    eventName,
    conversionValue,
    currencyCode,
    conversionCount,
  } = googleAdsConversionUploadSchema.parse(payload);
 
  try {
    const installedIntegration = await prisma.installedIntegration.findFirst({
      where: {
        integrationId: GOOGLE_ADS_INTEGRATION_ID,
        projectId: workspaceId,
      },
      select: {
        id: true,
        settings: true,
        credentials: true,
      },
    });
 
    if (!installedIntegration) {
      return {
        message: `Google Ads integration not installed for workspace ${workspaceId}. Skipping...`,
        status: "skipped",
      };
    }
 
    const settings = googleAdsSettingsSchema.parse(
      installedIntegration.settings ?? {},
    );
 
    const mappings =
      eventType === "lead" ? settings.leadMappings : settings.saleMappings;
    const mapping = resolveGoogleAdsConversionMapping({
      mappings,
      eventName,
    });
 
    if (!settings.customerId) {
      return {
        message: `Missing customerId. Skipping...`,
        status: "skipped",
      };
    }
 
    if (!mapping) {
      return {
        message: mappings.length
          ? `No ${eventType} conversion mapping matched event name "${eventName ?? ""}". Skipping...`
          : `Missing ${eventType} conversion mapping. Skipping...`,
        status: "skipped",
      };
    }
 
    const conversionAction = mapping.conversionAction;
 
    const googleClickId = extractGoogleAdsClickId(click.url);
 
    if (!googleClickId) {
      return {
        message: `No gclid/gbraid/wbraid found on click ${click.id}. Skipping...`,
        status: "skipped",
      };
    }

Copyconst token =
  await googleAdsOAuthProvider.getAccessToken(installedIntegration);
 
let loginCustomerId: string | null = null;
 
if (
  settings.loginCustomerId &&
  settings.loginCustomerId !== settings.customerId
) {
  loginCustomerId = settings.loginCustomerId;
}
 
const googleAdsApi = new GoogleAdsApi({
  accessToken: token.access_token,
  loginCustomerId,
  customerId: settings.customerId,
});
 
const maxRetries = 3;
 
for (let attempt = 0; attempt <= maxRetries; attempt++) {
  try {
    const response = await googleAdsApi.uploadClickConversion({
      customerId: settings.customerId,
      conversionAction,
      googleClickId,
      conversionDateTime,
      conversionValue,
      currencyCode,
      conversionCount,
      eventId,
    });
 
    return {
      message: `Uploaded ${eventType} conversion for workspace ${workspaceId} (requestId: ${response.requestId})`,
      status: "uploaded",
    };
  } catch (error) {
    if (attempt < maxRetries) {
      await sleep(1000 * Math.pow(2, attempt));
      continue;
    }
 
    throw error;
  }
}

Copyimport { COMMON_CORS_HEADERS } from "@/lib/api/cors";
import { parseRequestBody } from "@/lib/api/utils";
import {
  shopifyCheckoutCache,
  tryDispatchShopifyOrderJob,
} from "@/lib/integrations/shopify/checkout-cache";
import { getClickEvent } from "@/lib/tinybird";
import { ratelimit } from "@/lib/upstash";
import { LOCALHOST_IP } from "@dub/utils";
import { ipAddress, waitUntil } from "@vercel/functions";
import { NextResponse } from "next/server";
import * as z from "zod/v4";
 
const inputSchema = z.object({
  clickId: z.string().nullish(),
  checkoutToken: z.string().nullish(),
  shopDomain: z.string().nullish(),
});
 
// POST /api/shopify/pixel – Handle the Shopify Pixel events
export const POST = async (req: Request) => {
  const response = NextResponse.json("OK", {
    headers: COMMON_CORS_HEADERS,
  });

Copyif (!isLocalDev) {
  const ip = await getIP();
  const isAllowed = APPSFLYER_IP_RANGES.some((range) =>
    isIpInRange(ip, range),
  );
 
  if (!isAllowed) {
    throw new DubApiError({
      code: "forbidden",
      message: `IP address ${ip} is not allowed.`,
    });
  }
}

Copyconst querySchema = z.object({
  appId: z.string(),
  partnerEventId: z.string().nullish(),
  eventValue: z.string().nullish(),
});
…
queryParams = getSearchParams(req.url);
 
const { appId, partnerEventId } = querySchema.parse(queryParams);
 
if (!partnerEventId || !["lead", "sale"].includes(partnerEventId)) {
  return logAndRespond(
    `"${partnerEventId}" is not a valid partner event ID (accepted values: "lead", "sale"). Skipping...`,
  );
}

Copy// Find the installation
const installation = await prisma.installedIntegration.findFirst({
  where: {
    integrationId: APPSFLYER_INTEGRATION_ID,
    settings: {
      path: "$.appIds",
      array_contains: appId,
    },
  },
  select: {
    project: {
      select: {
        id: true,
        stripeConnectId: true,
        webhookEnabled: true,
      },
    },
  },
});

Copyif (!installation) {
  throw new DubApiError({
    code: "bad_request",
    message: `AppsFlyer integration is not configured for the appId: "${appId}". Did you add the app ID to the AppsFlyer integration settings?`,
  });
}

Copyif (partnerEventId === "lead") {
  const {
    clickId,
    eventName,
    customerExternalId,
    customerName,
    customerEmail,
    customerAvatar,
  } = trackLeadRequestSchema.parse(queryParams);
 
  const leadResponse = await trackLead({
    clickId,
    eventName,
    customerExternalId,
    customerName,
    customerEmail,
    customerAvatar,
    eventQuantity: undefined,
    mode: undefined,
    metadata: null,
    workspace,
    commissionSource: CommissionSource.appsflyer,
  });
 
  response = JSON.stringify(leadResponse);
}

Copy// Track sale event
else if (partnerEventId === "sale") {
  const amountInCents = appsflyerAmountToDubCents(queryParams.amount);
  const { eventName, customerExternalId, amount, currency, invoiceId } =
    trackSaleRequestSchema.parse({
      ...queryParams,
      ...(amountInCents !== undefined && { amount: amountInCents }),
    });
 
  const saleResponse = await trackSale({
    customerExternalId,
    amount,
    currency,
    eventName,
    paymentProcessor: "custom",
    invoiceId,
    leadEventName: undefined,
    metadata: null,
    workspace,
    commissionSource: CommissionSource.appsflyer,
  });
 
  response = JSON.stringify(saleResponse);
}

Copyimport { isLocalDev } from "@/lib/api/environment";
import { DubApiError, handleAndReturnErrorResponse } from "@/lib/api/errors";
import { getIP } from "@/lib/api/utils/get-ip";
import { normalizeWorkspaceId } from "@/lib/api/workspaces/workspace-id";
import { withAxiom } from "@/lib/axiom/server";
import { SINGULAR_IP_RANGES } from "@/lib/integrations/singular/singular-ip-ranges";
import { trackSingularLeadEvent } from "@/lib/integrations/singular/track-lead";
import { trackSingularSaleEvent } from "@/lib/integrations/singular/track-sale";
import { isIpInRange } from "@/lib/middleware/utils/is-ip-in-range";
import { prisma } from "@/lib/prisma";
import { getSearchParams } from "@dub/utils";
import { NextResponse } from "next/server";
import * as z from "zod/v4";
…
// GET /api/singular/webhook – listen to Postback events from Singular
export const GET = withAxiom(async (req) => {
  try {
    if (!isLocalDev) {
      const ip = await getIP();
      const isAllowed = SINGULAR_IP_RANGES.some((range) =>
        isIpInRange(ip, range),
      );
 
      if (!isAllowed) {
        throw new DubApiError({
          code: "forbidden",
          message: `IP address ${ip} is not allowed.`,
        });
      }
    }

Copyconst singularToDubEvent = {
  activated: "lead",
  sng_complete_registration: "lead",
  sng_subscribe: "sale",
  sng_ecommerce_purchase: "sale",
  __iap__: "sale", // In-app purchase
  "Copy GAID": "lead", // Singular Device Assist
  "copy IDFA": "lead", // Singular Device Assist
};
 
const supportedEvents = Object.keys(singularToDubEvent);
 
const querySchema = z.object({
  dub_workspace_id: z
    .string()
    .min(1, "dub_workspace_id is required")
    .describe(
      "The Singular advertiser's workspace ID on Dub (see https://d.to/id).",
    )
    .transform((v) => normalizeWorkspaceId(v)),
});
…
const queryParams = getSearchParams(req.url);
 
const { dub_workspace_id: workspaceId } = querySchema.parse(queryParams);
 
const { event_name: eventName } = queryParams;
 
if (!eventName) {
  throw new DubApiError({
    code: "bad_request",
    message: "event_name is required.",
  });
}
 
if (!supportedEvents.includes(eventName)) {
  console.error(
    `Event ${eventName} is not supported by Singular <> Dub integration.`,
  );
 
  return NextResponse.json("OK");
}

Copyconst workspace = await prisma.project.findUnique({
  where: {
    id: workspaceId,
  },
  select: {
    id: true,
    stripeConnectId: true,
    webhookEnabled: true,
  },
});
 
if (!workspace) {
  throw new DubApiError({
    code: "not_found",
    message: `Workspace ${workspaceId} not found.`,
  });
}

Copyconst dubEvent = singularToDubEvent[eventName];
 
delete queryParams.dub_workspace_id;
 
if (dubEvent === "lead") {
  await trackSingularLeadEvent({
    queryParams,
    workspace,
  });
} else if (dubEvent === "sale") {
  await trackSingularSaleEvent({
    queryParams,
    workspace,
  });
}

Copyexport const trackSingularLeadEvent = async ({
  queryParams,
  workspace,
}: {
  queryParams: Record<string, string>;
  workspace: Pick<WorkspaceProps, "id" | "stripeConnectId" | "webhookEnabled">;
}) => {
  const {
    dub_id: clickId,
    event_name: eventName,
    event_attributes: {
      customer_external_id: customerExternalId,
      customer_name: customerName,
      customer_email: customerEmail,
      customer_avatar: customerAvatar,
      event_quantity: eventQuantity,
      mode,
    },
  } = singularLeadEventSchema.parse(queryParams);
 
  return await trackLead({
    clickId,
    eventName,
    customerEmail,
    customerAvatar,
    customerExternalId,
    customerName,
    eventQuantity,
    mode,
    metadata: null,
    workspace,
    commissionSource: CommissionSource.singular,
  });
};

Copyexport interface PostbackPayload {
  eventId: string;
  event: PostbackTrigger;
  createdAt: string;
  data: unknown;
}
 
export abstract class PostbackAdapter {
  protected eventTransformers = new PostbackEventTransformers();
 
  constructor(protected postback: Postback) {
    this.registerEventTransformers();
  }
 
  protected abstract registerEventTransformers(): void;

Copyasync execute(payload: PostbackPayload) {
  const transformedPayload = this.eventTransformers.transform(payload);
 
  if (!transformedPayload) {
    return;
  }
 
  const searchParams = {
    postbackId: this.postback.id,
    eventId: payload.eventId,
    event: payload.event,
  };
 
  const callbackUrl = buildCallbackUrl(
    `${APP_DOMAIN_WITH_NGROK}/api/postbacks/callback`,
    searchParams,
  );

Copy  const signature = await createWebhookSignature(
    this.postback.secret,
    transformedPayload,
  );
 
  const response = await qstash.publishJSON({
    callback: callbackUrl.href,
    failureCallback: callbackUrl.href,
    url: this.postback.url,
    body: transformedPayload,
    headers: {
      "Dub-Signature": signature,
      "Upstash-Hide-Headers": "true",
    },
  });
 
  if (!response.messageId) {
    console.error("Failed to publish postback event to QStash", response);
  }
 
  if (process.env.NODE_ENV === "development") {
    console.debug("Published postback event to QStash", {
      ...response,
      payload: transformedPayload,
    });
  }
}

Copyimport { postbackEventInputSchemaTB } from "@/lib/postback/schemas";
import { tb } from "../tinybird/client";
 
export const recordPostbackEvent = tb.buildIngestEndpoint({
  datasource: "dub_postback_events",
  event: postbackEventInputSchemaTB.omit({ timestamp: true }),
});