Workflow Automation | blade47/dub Wiki | doc0

Public

Workflow Automation | blade47/dub Wiki | doc0

Model / Enum	Field Name	Type	Description
`WorkflowTrigger`	`partnerEnrolled`	Enum value	Triggered when a partner is enrolled (scheduled)
`WorkflowTrigger`	`partnerMetricsUpdated`	Enum value	Triggered when partner metrics are updated
`WorkflowTrigger`	`clickRecorded`	Enum value	Legacy trigger for recorded clicks
`WorkflowTrigger`	`commissionEarned`	Enum value	Legacy trigger for earned commissions
`Workflow`	`id`	String	Primary identifier
`Workflow`	`programId`	String	Foreign key to Program table
`Workflow`	`trigger`	WorkflowTrigger?	Optional trigger classification
`Workflow`	`triggerConditions`	Json	Serialized condition rules
`Workflow`	`actions`	Json	Serialized action payloads

export const workflowActionSchema = z.discriminatedUnion("type", [
  z.object({
    type: z.literal(WORKFLOW_ACTION_TYPES.AwardBounty),
    data: z.object({
      bountyId: z.string(),
    }),
  }),
  z.object({
    type: z.literal(WORKFLOW_ACTION_TYPES.SendCampaign),
    data: z.object({
      campaignId: z.string(),
    }),
  }),
  z.object({
    type: z.literal(WORKFLOW_ACTION_TYPES.MoveGroup),
    data: z.object({
      groupId: z.string(),
    }),
  }),
]);

export function checkWorkflowConditions({
  conditions,
  workflowType,
}: {
  conditions?: WorkflowCondition[] | null;
  workflowType: WorkflowType;
}): {
  valid: boolean;
  errors: string[];
} {
  if (!conditions || conditions.length === 0) {
    return { valid: true, errors: [] };
  }
  const attributes = WORKFLOW_TYPE_ATTRIBUTES[workflowType];
  const errors: string[] = [];
  for (let i = 0; i < conditions.length; i++) {
    const condition = conditions[i];
    if (!condition?.attribute) {
      errors.push(`Condition ${i + 1}: Please select an activity.`);
      continue;
    }
    const attributeDefinition = attributes[condition.attribute as keyof typeof attributes];
    if (!attributeDefinition) {
      errors.push(`Condition ${i + 1}: Invalid activity.`);
      continue;
    }
    const operatorDefinition = WORKFLOW_OPERATORS[condition.operator as keyof typeof WORKFLOW_OPERATORS];
    if (!operatorDefinition) {
      errors.push(`Condition ${i + 1}: Invalid operator.`);
      continue;
    }
    if (!(attributeDefinition.operators as readonly string[]).includes(condition.operator)) {
      errors.push(`Operator "${condition.operator}" is not valid for the activity "${condition.attribute}".`);
      continue;
    }
    if (attributeDefinition.inputType === "none") {
      continue;
    }
    if (condition.value == null) {
      errors.push(`Condition ${i + 1}: Please enter a value.`);
      continue;
    }
    try {
      operatorDefinition.validate(condition.value as any);
    } catch (error) {
      errors.push(`Condition ${i + 1}: ${error instanceof Error ? error.message : "Invalid value."}`);
    }
  }
  return { valid: errors.length === 0, errors };
}

Attribute Key	Label	Input Type	Allowed Operators	Data Requirements
`totalLeads`	total leads	number	`gte`	`partnerLinkStats`
`totalConversions`	total conversions	number	`gte`	`partnerLinkStats`
`totalSaleAmount`	total revenue	currency	`gte`	`partnerLinkStats`
`totalCommissions`	total commissions	currency	`gte`	`commissions`
`partnerEnrolledDays`	enrollment duration	dropdown	`gte`	none
`partnerJoined`	joins the program	none	`gte`	none
`partnerGroup`	group	group	`eq`, `ne`, `in`, `notIn`	none

Design Choice	Benefit	Cost
Discriminated unions for actions	Strict type safety and clear payload narrowing per action type	Requires explicit type mapping updates when adding new actions
Superset attribute definitions	Centralizes attribute rules across UI and API layers	Couples attribute keys to specific data fetching requirements
Standalone schema validation	Decouples condition checks from HTTP request lifecycles	Requires duplicate error context mapping for user-facing validation

Workflow Name	Route Path
`partner-approved-workflow`	`/api/workflows/partner-approved`
`merge-partner-accounts-workflow`	`/api/workflows/merge-partner-accounts`
`create-partner-commission-workflow`	`/api/workflows/create-partner-commission`
`reattribute-customer-workflow`	`/api/workflows/reattribute-customer`
`detach-discount-workflow`	`/api/workflows/detach-discount`

Design Choice	Benefit	Cost
Static workflow path mapping	Ensures strict compile-time validation between job names and HTTP endpoints	Requires manual registry updates when adding new workflow routes
Lazy commission aggregation	Avoids costly database queries when rules do not evaluate commission metrics	Introduces conditional branching complexity inside data loading promises
QStash chunked batching	Prevents payload size limit violations by splitting bulk dispatches	Requires error handling per chunk rather than individual transactions

Design Choice	Benefit	Cost
Serialized workflow execution (`parallelism: 1`)	Prevents race conditions and duplicate commission entries for the same partner	Serializes concurrent conversions for a single partner, potentially increasing queue latency
Historical commission lookups	Enables accurate recurring sale attribution and max-duration checks	Adds database query overhead to every lead and sale evaluation
Dual Redis caching for lead events	Supports rapid deduplication and immediate access before Tinybird ingestion	Requires managing separate cache keys with expiration windows

Parameter	Type	Default	Description
`bountyId`	`string` (required)	—	Unique identifier of the bounty being processed
`partnerIds`	`string[]` (optional)	`undefined`	Optional array of specific partner identifiers to filter scope
`page`	`number` (optional)	`0`	Pagination page offset for batching enrollment processing

Constant / Parameter	Type / Value	Description
`CUSTOMER_REATTRIBUTION_EVENTS_LIMIT`	`number` (`500`)	Maximum number of events processed during customer reattribution
`UNPAID_COMMISSION_STATUSES`	`readonly string[]` (`["pending", "hold", "processed"]`)	Commission statuses eligible for direct transfer during reattribution
`STATS_LOCK_TTL_SECONDS`	`number` (`86400`)	Time-to-live in seconds for statistics synchronization locks (24 hours)
`WorkflowRetryAfterError` (reingest)	`Error`, `"5s"` backoff	Exception thrown to retry event re-ingestion on failure
`WorkflowRetryAfterError` (delete)	`Error`, `"10s"` backoff	Exception thrown to retry Tinybird event deletion on failure

Design Choice	Benefit	Cost
Asynchronous multi-step workflow execution via Upstash	Ensures durable retries and decoupled execution for long-running analytics re-ingestion	Increases total completion latency due to step boundary serialization
Distributed run locking (`runOnce`) on clawback calculations	Prevents duplicate clawback generation and race conditions on financial adjustments	Requires persistent coordination state via Redis/Prisma
Event re-ingestion followed by old event deletion	Avoids temporary data loss windows during customer context switching	Briefly duplicates analytical event counts across partners during migration

failureFunction: async ({
  context,
  failStatus,
  failResponse,
  failHeaders,
}) => {
  logger.error("workflow.failed", {
    service: "qstash",
    event: "workflow.failed",
    workflowType: "detach-discount",
    workflowRunId: context.workflowRunId,
    discountId: context.requestPayload?.discountId,
    programId: context.requestPayload?.programId,
    failStatus,
    failResponse,
    failHeaders,
  });
 
  await logger.flush();
},

Trigger Identifier	Required Configuration	Behavior & Evaluation
`commissionEarned`	`trigger`	Calculates referral earnings as a percentage of the source commission's earnings.
`saleRecorded`	`trigger`	Calculates referral earnings as a percentage of the source commission's sale amount.
`partnerApproved`	`trigger`	Awards a flat amount (`amountInCents`) upon partner approval.
`commissionThreshold`	`trigger`, `commissionsThresholdInCents`	Aggregates earned sale commissions for the partner; grants reward if total meets or exceeds `commissionsThresholdInCents`.