Commission Engine

Copy// Skip invalid reward events (should never happen, but just in case)
if (event === "click" || event === "referral") {
  return logAndReturn({
    commission: null,
    outputLog: `"${event}" commissions are not created via this workflow. Skipping commission creation...`,
  });
}
 
// for lead / sale rewards, we need to check if the partner is eligible for commissions
if (
  ["lead", "sale"].includes(event) &&
  !COMMISSION_ELIGIBLE_ENROLLMENT_STATUSES.includes(programEnrollment.status)
) {
  return logAndReturn({
    commission: null,
    outputLog: `Partner ${partnerId} is not eligible for commissions (status: ${programEnrollment.status}) in program ${programId}, skipping ${event} commission creation...`,
  });
}

Copyexport const calculateSaleEarnings = ({
  reward,
  sale,
}: {
  reward: Pick<RewardProps, "type" | "amountInCents" | "amountInPercentage">;
  sale: Pick<Commission, "quantity" | "amount">;
}) => {
  if (!reward) {
    return 0;
  }
 
  const amount = getRewardAmount(reward);
 
  if (reward.type === "flat") {
    return sale.quantity * amount;
  } else if (reward.type === "percentage") {
    return Math.round((sale.amount * amount) / 100);
  }
 
  return 0;
};

Copy// for lead events, we need to check if the partner has already been issued a lead reward for this customer
if (event === "lead") {
  return logAndReturn({
    commission: null,
    outputLog: `Partner ${partnerId} has already been issued a lead reward for this customer ${customerId}, skipping commission creation...`,
  });
 
  // for sale rewards, we need to check if partner's reward was updated and different from the first commission's reward
  // we need to make sure it wasn't changed from one-time to recurring so we don't create a new commission
} else {
  if (
    firstCommission.rewardId &&
    firstCommission.rewardId !== reward.id
  ) {
    const originalReward = await prisma.reward.findUnique({
      where: {
        id: firstCommission.rewardId,
      },
      select: {
        id: true,
        maxDuration: true,
        modifiers: true,
      },
    });
 
    const originalMaxDuration = originalReward
      ? getRewardMaxDurationForContext({
          reward: originalReward,
          context,
        })
      : null;
 
    if (
      typeof originalMaxDuration === "number" &&
      originalMaxDuration === 0
    ) {
      return logAndReturn({
        commission: null,
        outputLog: `Partner ${partnerId} is only eligible for first-sale commissions based on the original reward ${originalReward?.id}, skipping commission creation...`,
      });
    }
  }
 
  // for sale rewards with a max duration, we need to check if the first commission is within the max duration
  // if it's beyond the max duration, we should not create a new commission
  if (typeof reward?.maxDuration === "number") {
    // One-time sale reward (maxDuration === 0)
    if (reward.maxDuration === 0) {
      return logAndReturn({
        commission: null,
        outputLog: `Partner ${partnerId} is only eligible for first-sale commissions, skipping commission creation...`,
      });
    }
 
    // Recurring sale reward (maxDuration > 0)
    else {
      const subscriptionDurationMonths = differenceInMonths(
        createdAt ?? new Date(), // account for custom commission creation date
        firstCommission.createdAt,
      );
 
      if (subscriptionDurationMonths >= reward.maxDuration) {
        return logAndReturn({
          commission: null,

Copyexport const POST = withWorkspace(
  async ({ workspace, session, req }) => {
    const programId = getDefaultProgramIdOrThrow(workspace);
 
    const body = createManualCommissionBodySchema.parse(
      await parseRequestBody(req),
    );
 
    console.time("createManualCommissions");
 
    await createManualCommissions({
      ...body,
      workspace,
      programId,
      user: session.user,
    });
 
    console.timeEnd("createManualCommissions");
 
    const isClawback = body.type === "custom" && body.amount < 0;
 
    const response = createCommissionResponseSchema.parse({
      success: true,
      message: isClawback
        ? "A clawback has been queued for the partner!"
        : "Your commissions are being created and will appear shortly.",
    });
 
    return NextResponse.json(response, {
      status: 202,
    });
  },

Copyif (type === "sale") {
  const { importStripeInvoices, sale } = args;
 
  if (importStripeInvoices) {
    if (!workspace.stripeConnectId) {
      throw new DubApiError({
        code: "bad_request",
        message: `Your workspace isn't connected to Stripe yet. Please install the Stripe integration to continue: https://app.dub.co/${workspace.slug}/settings/integrations/stripe`,
      });
    }
 
    if (targetCustomer && !targetCustomer.stripeCustomerId) {
      throw new DubApiError({
        code: "bad_request",
        message: `Customer ${targetCustomer.id} does not have a Stripe Customer ID configured. Please update the customer record at https://app.dub.co/${workspace.slug}/program/customers/${targetCustomer.id}`,
      });
    }
  }
 
  const invoiceId = sale?.invoiceId ?? args.invoiceId;
 
  if (invoiceId) {
    const commission = await prisma.commission.findUnique({
      where: {
        invoiceId_programId: {
          invoiceId,
          programId,
        },
      },
      select: {
        id: true,
      },
    });
 
    if (commission) {
      throw new DubApiError({
        code: "conflict",
        message: `There is already a commission for the invoice ${invoiceId}.`,
      });
    }
  }
}

Copyconst invoicesToProcess = invoices.data.filter((invoice) => {
  // skip $0 invoices
  if (invoice.amount_paid === 0) {
    return false;
  }
  const existingCommission = existingCommissions.find(
    (commission) =>
      commission.invoiceId === invoice.id ||
      (commission.amount === invoice.amount_paid &&
        new Date(commission.createdAt).getTime() >=
          new Date(invoice.created * 1000).getTime() - 60 * 60 * 1000 &&
        new Date(commission.createdAt).getTime() <=
          new Date(invoice.created * 1000).getTime() + 60 * 60 * 1000),
  );
  return !existingCommission;
});

Copyif (partnerId && referredByPartnerId && partnerId === referredByPartnerId) {
  console.log(
    `Skipping referral commission creation for self-referral (partner ${partnerId}).`,
  );
  return null;
}

Copyif (
  [NETWORK_PROGRAM_ID, ACME_PROGRAM_ID, DEMO_PROGRAM_ID].includes(
    payout.programId,
  )
) {
  console.error(
    `Payout ${payout.id} is from Network, Acme, or Demo program, skipping...`,
  );
  return null;
}
 
const payoutFeeEarned = Math.floor(payout.amount * 0.03); // assumes 3% average payout fee for all payouts

Copyif (firstCommission) {
  const durationMonths = differenceInMonths(
    new Date(),
    firstCommission.createdAt,
  );
 
  if (durationMonths >= NETWORK_REFERRAL_REWARD.maxDuration) {
    console.log(
      `Referrer partner ${partner.referredByPartnerId} has reached max duration for network bonus.`,
    );
    return null;
  }
}

Copyif (error.code === "P2002") {
  console.log(
    `Referral commission already exists for invoiceId ${commissionData.invoiceId}, skipping creation.`,
  );
  return null;
}

Copyif (plan.shouldClearIdentifiers && plan.paidCommissionIds.length > 0) {
  await prisma.commission.updateMany({
    where: {
      id: { in: plan.paidCommissionIds },
    },
    data: {
      eventId: null,
      invoiceId: null,
    },
  });
}
 
await runOnce({
  key: `reattribute-customer:${oldCustomerId}:clawback`,
  fn: async () => {
    const existingClawback = await prisma.commission.findFirst({
      where: {
        customerId: oldCustomerId,
        partnerId: oldPartnerId,
        programId,
        type: "custom",
        description: "tracking_error",
        earnings: { lt: 0 },
      },
      select: { id: true },
    });
 
    if (existingClawback) {
      return;
    }
 
    await queuePartnerCommissionCreation({
      event: "custom",
      partnerId: oldPartnerId,
      programId,
      customerId: oldCustomerId,
      amount: -plan.paidEarnings,
      quantity: 1,
      description: "tracking_error",
      skipWorkflow: true,
    });
  },
});

Copyconst [newEvents, existingNewCommissions] = await Promise.all([
  getCustomerReattributeEvents(newCustomerId),
  prisma.commission.findMany({
    where: {
      customerId: newCustomerId,
      partnerId: newPartnerId,
      type: { in: ["lead", "sale"] },
    },
    select: {
      type: true,
      invoiceId: true,
    },
  }),
]);
 
const hasExistingLead = existingNewCommissions.some(
  (commission) => commission.type === "lead",
);
 
const leadEvent =
  plan.hasPaidLead && !hasExistingLead
    ? newEvents.find((event) => event.event === "lead") ?? null
    : null;
 
const paidInvoiceIds = new Set(plan.paidInvoiceIds);
const existingInvoiceIds = new Set(
  existingNewCommissions
    .map((commission) => commission.invoiceId)
    .filter((invoiceId): invoiceId is string => Boolean(invoiceId)),
);
const existingInvoicelessCount = existingNewCommissions.filter(
  (commission) => commission.type === "sale" && !commission.invoiceId,
).length;
 
const saleEvents = [
  ...newEvents.filter(
    (event) =>
      event.event === "sale" &&
      event.invoice_id &&
      paidInvoiceIds.has(event.invoice_id) &&
      !existingInvoiceIds.has(event.invoice_id),
  ),
  ...newEvents
    .filter((event) => event.event === "sale" && !event.invoice_id)
    .slice(
      0,
      Math.max(0, plan.paidInvoicelessCount - existingInvoicelessCount),
    ),
];

Copyexport async function trackCommissionStatusUpdate({
  commissions,
  newStatus,
  ...baseInput
}: {
  commissions: Pick<Commission, "id" | "amount" | "earnings" | "status">[];
  newStatus: CommissionStatus;
} & Pick<TrackActivityLogInput, "workspaceId" | "programId" | "userId">) {
  if (commissions.length === 0) {
    return;
  }
 
  return trackCommissionActivityLog({
    ...baseInput,
    old: commissions,
    new: commissions.map((commission) => ({
      ...commission,
      status: newStatus,
    })),
  });
}

Copyexport const syncTotalCommissions = async ({
  partnerId,
  programId,
}: {
  partnerId: string;
  programId: string;
}) => {
  try {
    return await publishPartnerActivityEvent({
      programId,
      partnerId,
      eventType: "commission",
      timestamp: new Date().toISOString(),
    });
  } catch (error) {
    console.error(
      `[syncTotalCommissionsError]: Failed to sync total commissions for partner ${partnerId} in program ${programId}, falling back to direct database update...`,
      error,
    );
 
    return await aggregateAndUpdateTotalCommissions({ partnerId, programId });
  }
};

Copyexport async function getCommissions(filters: CommissionsFilters) {
  const {
    invoiceId,
    programId,
    partnerId,
    status,
    type,
    customerId,
    payoutId,
    bountySubmissionId,
    groupId,
    partnerTagId,
    fraudEventGroupId,
    start,
    end,
    interval,
    timezone,
    startingAfter,
    endingBefore,
    query,
  } = filters;
 
  // Metadata filter
  const parsedMetadataQuery = parseCommissionMetadataQuery(query);
  const metadataWhere = buildCommissionMetadataWhere(parsedMetadataQuery);
 
  // InvoiceId is unique within a program
  if (invoiceId) {
    return await prisma.commission.findMany({
      where: {
        invoiceId,
        programId,
        ...metadataWhere,
      },
      include: {
        ...commissionIncludes,
      },
    });
  }
 
  const paginationQuery = buildPaginationQuery(filters);
 
  // Validate the provided cursor ID
  const cursorId = startingAfter || endingBefore;
 
  if (cursorId) {
    const commission = await prisma.commission.findUnique({
      where: {
        id: cursorId,
      },
      select: {
        id: true,
        programId: true,
      },
    });

Copyexport async function getCommissionsTimeseries({
  programId,
  startDate,
  endDate,
  granularity,
  timezone,
}: {
  programId?: string;
  startDate: Date;
  endDate: Date;
  granularity: string;
  timezone: string;
}) {
  const { dateFormat, dateIncrement, startFunction, formatString } =
    sqlGranularityMap[granularity];
 
  const commissions = await prisma.$queryRaw<Commission[]>`
        SELECT 
          DATE_FORMAT(CONVERT_TZ(createdAt, "UTC", ${timezone || "UTC"}), ${dateFormat}) AS start, 
          SUM(earnings) AS commissions
        FROM Commission
        WHERE 
          createdAt >= ${startDate}
          AND createdAt < ${endDate}
          AND status IN ("pending", "processed", "paid")
          AND ${programId ? Prisma.sql`programId = ${programId}` : Prisma.sql`programId NOT IN (${ACME_PROGRAM_ID}, ${DEMO_PROGRAM_ID})`}
        GROUP BY start
        ORDER BY start ASC;`;