Examples
Real-world feature patterns using the code-first API. For conditionals, loops, and switch (e.g. branchOverrides, recordInput, recordScenarios), see Building Features - Conditionals, loops, and switch.
E-Commerce Order Fulfillment
A complete order processing feature with payment, inventory, and notifications. Because the handler returns early when validation fails, use branchOverrides so the compiler records the payment and later steps; at runtime those steps run only when the condition holds.
- TypeScript
- Java
- Go
- .NET
await ductape.features.define({
tag: 'order-fulfillment',
name: 'Order Fulfillment',
description: 'Process customer orders from payment to delivery',
options: {
timeout: 600000, // 10 minutes
rollback_strategy: 'reverse_all',
},
signals: {
'cancel-order': { input: { reason: 'string' } },
},
branchOverrides: { 'validate-order': { available: true } },
handler: async (ctx) => {
ctx.log.info('Starting order fulfillment', { orderId: ctx.input.orderId });
// Step 1: Validate order and check inventory
const validation = await ctx.step('validate-order', async () => {
const inventory = await ctx.api.run({
app: 'inventory-api',
event: 'check-availability',
input: { body: { items: ctx.input.items } },
});
return {
available: inventory.allAvailable,
unavailableItems: inventory.unavailableItems || [],
};
});
if (!validation.available) {
return {
success: false,
reason: 'items_unavailable',
unavailableItems: validation.unavailableItems,
};
}
// Step 2: Calculate totals and apply discounts
const pricing = await ctx.step('calculate-pricing', async () => {
const subtotal = ctx.input.items.reduce(
(sum, item) => sum + item.price * item.quantity,
0
);
let discount = 0;
if (ctx.input.couponCode) {
const coupon = await ctx.api.run({
app: 'promotions-api',
event: 'validate-coupon',
input: { body: { code: ctx.input.couponCode } },
});
if (coupon.valid) {
discount = coupon.discountAmount;
}
}
const tax = (subtotal - discount) * 0.08; // 8% tax
const total = subtotal - discount + tax;
return { subtotal, discount, tax, total };
});
// Step 3: Process payment (with rollback)
const payment = await ctx.step(
'process-payment',
async () => {
return ctx.api.run({
app: 'stripe',
event: 'create-charge',
input: {
body: {
amount: Math.round(pricing.total * 100), // cents
currency: 'usd',
customer: ctx.input.customerId,
metadata: { orderId: ctx.input.orderId },
},
headers: {
'Idempotency-Key': `order-${ctx.input.orderId}-payment`,
},
},
retries: 3,
timeout: 30000,
});
},
async (result) => {
// Rollback: Refund the charge
await ctx.api.run({
app: 'stripe',
event: 'create-refund',
input: {
body: { charge: result.id, reason: 'order_cancelled' },
},
});
ctx.log.info('Payment refunded', { chargeId: result.id });
}
);
// Checkpoint after payment
await ctx.checkpoint('payment-complete', {
chargeId: payment.id,
amount: pricing.total,
});
// Step 4: Reserve inventory (with rollback)
await ctx.step(
'reserve-inventory',
async () => {
return ctx.api.run({
app: 'inventory-api',
event: 'reserve-items',
input: {
body: {
orderId: ctx.input.orderId,
items: ctx.input.items,
},
},
});
},
async () => {
// Rollback: Release reserved items
await ctx.api.run({
app: 'inventory-api',
event: 'release-items',
input: { body: { orderId: ctx.input.orderId } },
});
}
);
// Step 5: Create order record
const order = await ctx.step('create-order-record', async () => {
return ctx.database.insert({
database: 'orders-db',
event: 'create-order',
data: {
order_id: ctx.input.orderId,
customer_id: ctx.input.customerId,
items: ctx.input.items,
subtotal: pricing.subtotal,
discount: pricing.discount,
tax: pricing.tax,
total: pricing.total,
payment_id: payment.id,
status: 'confirmed',
created_at: new Date().toISOString(),
},
});
});
// Step 6: Send confirmation email (non-critical)
await ctx.step(
'send-confirmation-email',
async () => {
await ctx.notification.email({
notification: 'transactional',
event: 'order-confirmed',
recipients: [ctx.input.email],
subject: { orderId: ctx.input.orderId },
template: {
orderId: ctx.input.orderId,
items: ctx.input.items,
total: pricing.total,
estimatedDelivery: '3-5 business days',
},
});
},
null,
{ allow_fail: true }
);
// Step 7: Queue for warehouse fulfillment
await ctx.step('queue-fulfillment', async () => {
await ctx.messaging.produce({
event: 'warehouse-events:new-order',
message: {
orderId: ctx.input.orderId,
items: ctx.input.items,
shippingAddress: ctx.input.shippingAddress,
priority: ctx.input.isPrime ? 'high' : 'normal',
},
});
});
ctx.log.info('Order fulfilled successfully', {
orderId: ctx.input.orderId,
chargeId: payment.id,
});
return {
success: true,
orderId: ctx.input.orderId,
orderRecordId: order.id,
chargeId: payment.id,
total: pricing.total,
};
},
});
ductape.features.define(Map.of(
"tag", "order-fulfillment",
"name", "Order Fulfillment",
"description", "Process customer orders from payment to delivery",
options: Map.of(
"timeout", 600000, // 10 minutes
"rollback_strategy", "reverse_all"
),
signals: Map.of(
'cancel-order': Map.of( input: Map.of( "reason", "string" ) )
),
branchOverrides: Map.of( 'validate-order': Map.of( "available", true ) ),
handler: async (ctx) => Map.of(
ctx.log.info('Starting order fulfillment', Map.of( orderId: ctx.input.orderId ));
// Step 1: Validate order and check inventory
Map<String, Object> validation = ctx.step('validate-order', async () => Map.of(
Map<String, Object> inventory = ctx.api.run(Map.of(
"app", "inventory-api",
"event", "check-availability",
input: Map.of( body: Map.of( items: ctx.input.items ) )
));
return Map.of(
available: inventory.allAvailable,
unavailableItems: inventory.unavailableItems || []
);
));
if (!validation.available) Map.of(
return Map.of(
"success", false,
"reason", "items_unavailable",
unavailableItems: validation.unavailableItems
);
)
// Step 2: Calculate totals and apply discounts
Map<String, Object> pricing = ctx.step('calculate-pricing', async () => Map.of(
Map<String, Object> subtotal = ctx.input.items.reduce(
(sum, item) => sum + item.price * item.quantity,
0
);
Map<String, Object> discount = 0;
if (ctx.input.couponCode) Map.of(
Map<String, Object> coupon = ctx.api.run(Map.of(
"app", "promotions-api",
"event", "validate-coupon",
input: Map.of( body: Map.of( code: ctx.input.couponCode ) )
));
if (coupon.valid) Map.of(
discount = coupon.discountAmount;
)
)
Map<String, Object> tax = (subtotal - discount) * 0.08; // 8% tax
Map<String, Object> total = subtotal - discount + tax;
return Map.of( subtotal, discount, tax, total );
));
// Step 3: Process payment (with rollback)
Map<String, Object> payment = ctx.step(
'process-payment',
async () => Map.of(
return ctx.api.run(Map.of(
"app", "stripe",
"event", "create-charge",
input: Map.of(
body: Map.of(
amount: Math.round(pricing.total * 100), // cents
"currency", "usd",
customer: ctx.input.customerId,
metadata: Map.of( orderId: ctx.input.orderId )
),
headers: Map.of(
'Idempotency-Key': `order-$Map.of(ctx.input.orderId)-payment`
)
),
"retries", 3,
"timeout", 30000
));
),
async (result) => Map.of(
// Rollback: Refund the charge
ctx.api.run(Map.of(
"app", "stripe",
"event", "create-refund",
input: Map.of(
body: Map.of( charge: result.id, "reason", "order_cancelled" )
)
));
ctx.log.info('Payment refunded', Map.of( chargeId: result.id ));
)
);
// Checkpoint after payment
ctx.checkpoint('payment-complete', Map.of(
chargeId: payment.id,
amount: pricing.total
));
// Step 4: Reserve inventory (with rollback)
ctx.step(
'reserve-inventory',
async () => Map.of(
return ctx.api.run(Map.of(
"app", "inventory-api",
"event", "reserve-items",
input: Map.of(
body: Map.of(
orderId: ctx.input.orderId,
items: ctx.input.items
)
)
));
),
async () => Map.of(
// Rollback: Release reserved items
ctx.api.run(Map.of(
"app", "inventory-api",
"event", "release-items",
input: Map.of( body: Map.of( orderId: ctx.input.orderId ) )
));
)
);
// Step 5: Create order record
Map<String, Object> order = ctx.step('create-order-record', async () => Map.of(
return ctx.database.insert(Map.of(
"database", "orders-db",
"event", "create-order",
data: Map.of(
order_id: ctx.input.orderId,
customer_id: ctx.input.customerId,
items: ctx.input.items,
subtotal: pricing.subtotal,
discount: pricing.discount,
tax: pricing.tax,
total: pricing.total,
payment_id: payment.id,
"status", "confirmed",
created_at: Instant.now().toISOString()
)
));
));
// Step 6: Send confirmation email (non-critical)
ctx.step(
'send-confirmation-email',
async () => Map.of(
ctx.notification.email(Map.of(
"notification", "transactional",
"event", "order-confirmed",
recipients: [ctx.input.email],
subject: Map.of( orderId: ctx.input.orderId ),
template: Map.of(
orderId: ctx.input.orderId,
items: ctx.input.items,
total: pricing.total,
"estimatedDelivery", "3-5 business days"
)
));
),
null,
Map.of( "allow_fail", true )
);
// Step 7: Queue for warehouse fulfillment
ctx.step('queue-fulfillment', async () => Map.of(
ctx.messaging.produce(Map.of(
"event", "warehouse-events:new-order",
message: Map.of(
orderId: ctx.input.orderId,
items: ctx.input.items,
shippingAddress: ctx.input.shippingAddress,
priority: ctx.input.isPrime ? 'high' : 'normal'
)
));
));
ctx.log.info('Order fulfilled successfully', Map.of(
orderId: ctx.input.orderId,
chargeId: payment.id
));
return Map.of(
"success", true,
orderId: ctx.input.orderId,
orderRecordId: order.id,
chargeId: payment.id,
total: pricing.total
);
)
));
client.features.define({
"tag": "order-fulfillment",
"name": "Order Fulfillment",
"description": "Process customer orders from payment to delivery",
options: {
"timeout": 600000, // 10 minutes
"rollback_strategy": "reverse_all",
},
signals: {
'cancel-order': { input: { "reason": "string" } },
},
branchOverrides: { 'validate-order': { "available": true } },
handler: async (ctx) => {
ctx.log.info('Starting order fulfillment', { orderId: ctx.input.orderId });
// Step 1: Validate order and check inventory
validation := ctx.step('validate-order', async () => {
inventory := ctx.api.run({
"app": "inventory-api",
"event": "check-availability",
input: { body: { items: ctx.input.items } },
});
return {
available: inventory.allAvailable,
unavailableItems: inventory.unavailableItems || [],
};
});
if (!validation.available) {
return {
"success": false,
"reason": "items_unavailable",
unavailableItems: validation.unavailableItems,
};
}
// Step 2: Calculate totals and apply discounts
pricing := ctx.step('calculate-pricing', async () => {
subtotal := ctx.input.items.reduce(
(sum, item) => sum + item.price * item.quantity,
0
);
discount := 0;
if (ctx.input.couponCode) {
coupon := ctx.api.run({
"app": "promotions-api",
"event": "validate-coupon",
input: { body: { code: ctx.input.couponCode } },
});
if (coupon.valid) {
discount = coupon.discountAmount;
}
}
tax := (subtotal - discount) * 0.08; // 8% tax
total := subtotal - discount + tax;
return { subtotal, discount, tax, total };
});
// Step 3: Process payment (with rollback)
payment := ctx.step(
'process-payment',
async () => {
return ctx.api.run({
"app": "stripe",
"event": "create-charge",
input: {
body: {
amount: Math.round(pricing.total * 100), // cents
"currency": "usd",
customer: ctx.input.customerId,
metadata: { orderId: ctx.input.orderId },
},
headers: {
'Idempotency-Key': `order-${ctx.input.orderId}-payment`,
},
},
"retries": 3,
"timeout": 30000,
});
},
async (result) => {
// Rollback: Refund the charge
ctx.api.run({
"app": "stripe",
"event": "create-refund",
input: {
body: { charge: result.id, "reason": "order_cancelled" },
},
});
ctx.log.info('Payment refunded', { chargeId: result.id });
}
);
// Checkpoint after payment
ctx.checkpoint('payment-complete', {
chargeId: payment.id,
amount: pricing.total,
});
// Step 4: Reserve inventory (with rollback)
ctx.step(
'reserve-inventory',
async () => {
return ctx.api.run({
"app": "inventory-api",
"event": "reserve-items",
input: {
body: {
orderId: ctx.input.orderId,
items: ctx.input.items,
},
},
});
},
async () => {
// Rollback: Release reserved items
ctx.api.run({
"app": "inventory-api",
"event": "release-items",
input: { body: { orderId: ctx.input.orderId } },
});
}
);
// Step 5: Create order record
order := ctx.step('create-order-record', async () => {
return ctx.database.insert({
"database": "orders-db",
"event": "create-order",
data: {
order_id: ctx.input.orderId,
customer_id: ctx.input.customerId,
items: ctx.input.items,
subtotal: pricing.subtotal,
discount: pricing.discount,
tax: pricing.tax,
total: pricing.total,
payment_id: payment.id,
"status": "confirmed",
created_at: new Date().toISOString(),
},
});
});
// Step 6: Send confirmation email (non-critical)
ctx.step(
'send-confirmation-email',
async () => {
ctx.notification.email({
"notification": "transactional",
"event": "order-confirmed",
recipients: [ctx.input.email],
subject: { orderId: ctx.input.orderId },
template: {
orderId: ctx.input.orderId,
items: ctx.input.items,
total: pricing.total,
"estimatedDelivery": "3-5 business days",
},
});
},
null,
{ "allow_fail": true }
);
// Step 7: Queue for warehouse fulfillment
ctx.step('queue-fulfillment', async () => {
ctx.messaging.produce({
"event": "warehouse-events:new-order",
message: {
orderId: ctx.input.orderId,
items: ctx.input.items,
shippingAddress: ctx.input.shippingAddress,
priority: ctx.input.isPrime ? 'high' : 'normal',
},
});
});
ctx.log.info('Order fulfilled successfully', {
orderId: ctx.input.orderId,
chargeId: payment.id,
});
return {
"success": true,
orderId: ctx.input.orderId,
orderRecordId: order.id,
chargeId: payment.id,
total: pricing.total,
};
},
});
await ductape.features.define({
["tag"] = "order-fulfillment",
["name"] = "Order Fulfillment",
["description"] = "Process customer orders from payment to delivery",
options: {
["timeout"] = 600000, // 10 minutes
["rollback_strategy"] = "reverse_all",
},
signals: {
'cancel-order': { input: { ["reason"] = "string" } },
},
branchOverrides: { 'validate-order': { ["available"] = true } },
handler: async (ctx) => {
ctx.log.info('Starting order fulfillment', { orderId: ctx.input.orderId });
// Step 1: Validate order and check inventory
var validation = await ctx.step('validate-order', async () => {
var inventory = await ctx.api.run({
["app"] = "inventory-api",
["event"] = "check-availability",
input: { body: { items: ctx.input.items } },
});
return {
available: inventory.allAvailable,
unavailableItems: inventory.unavailableItems || [],
};
});
if (!validation.available) {
return {
["success"] = false,
["reason"] = "items_unavailable",
unavailableItems: validation.unavailableItems,
};
}
// Step 2: Calculate totals and apply discounts
var pricing = await ctx.step('calculate-pricing', async () => {
var subtotal = ctx.input.items.reduce(
(sum, item) => sum + item.price * item.quantity,
0
);
var discount = 0;
if (ctx.input.couponCode) {
var coupon = await ctx.api.run({
["app"] = "promotions-api",
["event"] = "validate-coupon",
input: { body: { code: ctx.input.couponCode } },
});
if (coupon.valid) {
discount = coupon.discountAmount;
}
}
var tax = (subtotal - discount) * 0.08; // 8% tax
var total = subtotal - discount + tax;
return { subtotal, discount, tax, total };
});
// Step 3: Process payment (with rollback)
var payment = await ctx.step(
'process-payment',
async () => {
return ctx.api.run({
["app"] = "stripe",
["event"] = "create-charge",
input: {
body: {
amount: Math.round(pricing.total * 100), // cents
["currency"] = "usd",
customer: ctx.input.customerId,
metadata: { orderId: ctx.input.orderId },
},
headers: {
'Idempotency-Key': `order-${ctx.input.orderId}-payment`,
},
},
["retries"] = 3,
["timeout"] = 30000,
});
},
async (result) => {
// Rollback: Refund the charge
await ctx.api.run({
["app"] = "stripe",
["event"] = "create-refund",
input: {
body: { charge: result.id, ["reason"] = "order_cancelled" },
},
});
ctx.log.info('Payment refunded', { chargeId: result.id });
}
);
// Checkpoint after payment
await ctx.checkpoint('payment-complete', {
chargeId: payment.id,
amount: pricing.total,
});
// Step 4: Reserve inventory (with rollback)
await ctx.step(
'reserve-inventory',
async () => {
return ctx.api.run({
["app"] = "inventory-api",
["event"] = "reserve-items",
input: {
body: {
orderId: ctx.input.orderId,
items: ctx.input.items,
},
},
});
},
async () => {
// Rollback: Release reserved items
await ctx.api.run({
["app"] = "inventory-api",
["event"] = "release-items",
input: { body: { orderId: ctx.input.orderId } },
});
}
);
// Step 5: Create order record
var order = await ctx.step('create-order-record', async () => {
return ctx.database.insert({
["database"] = "orders-db",
["event"] = "create-order",
data: {
order_id: ctx.input.orderId,
customer_id: ctx.input.customerId,
items: ctx.input.items,
subtotal: pricing.subtotal,
discount: pricing.discount,
tax: pricing.tax,
total: pricing.total,
payment_id: payment.id,
["status"] = "confirmed",
created_at: DateTime.UtcNow.toISOString(),
},
});
});
// Step 6: Send confirmation email (non-critical)
await ctx.step(
'send-confirmation-email',
async () => {
await ctx.notification.email({
["notification"] = "transactional",
["event"] = "order-confirmed",
recipients: [ctx.input.email],
subject: { orderId: ctx.input.orderId },
template: {
orderId: ctx.input.orderId,
items: ctx.input.items,
total: pricing.total,
["estimatedDelivery"] = "3-5 business days",
},
});
},
null,
{ ["allow_fail"] = true }
);
// Step 7: Queue for warehouse fulfillment
await ctx.step('queue-fulfillment', async () => {
await ctx.messaging.produce({
["event"] = "warehouse-events:new-order",
message: {
orderId: ctx.input.orderId,
items: ctx.input.items,
shippingAddress: ctx.input.shippingAddress,
priority: ctx.input.isPrime ? 'high' : 'normal',
},
});
});
ctx.log.info('Order fulfilled successfully', {
orderId: ctx.input.orderId,
chargeId: payment.id,
});
return {
["success"] = true,
orderId: ctx.input.orderId,
orderRecordId: order.id,
chargeId: payment.id,
total: pricing.total,
};
},
});
User Onboarding
Multi-step user registration with email verification:
- TypeScript
- Java
- Go
- .NET
await ductape.features.define({
tag: 'user-onboarding',
name: 'User Onboarding',
signals: {
'email-verified': { input: { code: 'string' } },
},
handler: async (ctx) => {
// Step 1: Create user account
const user = await ctx.step(
'create-user',
async () => {
return ctx.database.insert({
database: 'users-db',
event: 'create-user',
data: {
email: ctx.input.email,
name: ctx.input.name,
password_hash: ctx.input.passwordHash,
status: 'pending_verification',
created_at: new Date().toISOString(),
},
});
},
async (result) => {
// Rollback: Delete user if onboarding fails
await ctx.database.delete({
database: 'users-db',
event: 'delete-user',
where: { id: result.id },
});
}
);
// Step 2: Generate verification code
const verificationCode = Math.random().toString(36).substring(2, 8).toUpperCase();
ctx.setState('verificationCode', verificationCode);
// Step 3: Send verification email
await ctx.step('send-verification', async () => {
await ctx.notification.email({
notification: 'transactional',
event: 'email-verification',
recipients: [ctx.input.email],
subject: { appName: 'MyApp' },
template: {
name: ctx.input.name,
code: verificationCode,
expiresIn: '24 hours',
},
});
});
// Step 4: Wait for email verification
const verification = await ctx.waitForSignal('email-verified', {
timeout: '24h',
});
// Validate the code
if (verification.code !== ctx.getState('verificationCode')) {
await ctx.triggerRollback('Invalid verification code');
return { success: false, reason: 'invalid_code' };
}
// Step 5: Activate user account
await ctx.step('activate-user', async () => {
await ctx.database.update({
database: 'users-db',
event: 'update-user',
where: { id: user.id },
data: {
status: 'active',
verified_at: new Date().toISOString(),
},
});
});
// Step 6: Create user profile
await ctx.step('create-profile', async () => {
await ctx.database.insert({
database: 'profiles-db',
event: 'create-profile',
data: {
user_id: user.id,
display_name: ctx.input.name,
avatar_url: null,
bio: null,
},
});
});
// Step 7: Send welcome email (non-critical)
await ctx.step(
'send-welcome',
async () => {
await ctx.notification.email({
notification: 'transactional',
event: 'welcome',
recipients: [ctx.input.email],
template: {
name: ctx.input.name,
loginUrl: 'https://myapp.com/login',
},
});
},
null,
{ allow_fail: true }
);
return {
success: true,
userId: user.id,
email: ctx.input.email,
};
},
});
ductape.features.define(Map.of(
"tag", "user-onboarding",
"name", "User Onboarding",
signals: Map.of(
'email-verified': Map.of( input: Map.of( "code", "string" ) )
),
handler: async (ctx) => Map.of(
// Step 1: Create user account
Map<String, Object> user = ctx.step(
'create-user',
async () => Map.of(
return ctx.database.insert(Map.of(
"database", "users-db",
"event", "create-user",
data: Map.of(
email: ctx.input.email,
name: ctx.input.name,
password_hash: ctx.input.passwordHash,
"status", "pending_verification",
created_at: Instant.now().toISOString()
)
));
),
async (result) => Map.of(
// Rollback: Delete user if onboarding fails
ctx.database.delete(Map.of(
"database", "users-db",
"event", "delete-user",
where: Map.of( id: result.id )
));
)
);
// Step 2: Generate verification code
Map<String, Object> verificationCode = Math.random().toString(36).substring(2, 8).toUpperCase();
ctx.setState('verificationCode', verificationCode);
// Step 3: Send verification email
ctx.step('send-verification', async () => Map.of(
ctx.notification.email(Map.of(
"notification", "transactional",
"event", "email-verification",
recipients: [ctx.input.email],
subject: Map.of( "appName", "MyApp" ),
template: Map.of(
name: ctx.input.name,
code: verificationCode,
"expiresIn", "24 hours"
)
));
));
// Step 4: Wait for email verification
Map<String, Object> verification = ctx.waitForSignal('email-verified', Map.of(
"timeout", "24h"
));
// Validate the code
if (verification.code !== ctx.getState('verificationCode')) Map.of(
ctx.triggerRollback('Invalid verification code');
return Map.of( "success", false, "reason", "invalid_code" );
)
// Step 5: Activate user account
ctx.step('activate-user', async () => Map.of(
ctx.database.update(Map.of(
"database", "users-db",
"event", "update-user",
where: Map.of( id: user.id ),
data: Map.of(
"status", "active",
verified_at: Instant.now().toISOString()
)
));
));
// Step 6: Create user profile
ctx.step('create-profile', async () => Map.of(
ctx.database.insert(Map.of(
"database", "profiles-db",
"event", "create-profile",
data: Map.of(
user_id: user.id,
display_name: ctx.input.name,
avatar_url: null,
bio: null
)
));
));
// Step 7: Send welcome email (non-critical)
ctx.step(
'send-welcome',
async () => Map.of(
ctx.notification.email(Map.of(
"notification", "transactional",
"event", "welcome",
recipients: [ctx.input.email],
template: Map.of(
name: ctx.input.name,
"loginUrl", "https://myapp.com/login"
)
));
),
null,
Map.of( "allow_fail", true )
);
return Map.of(
"success", true,
userId: user.id,
email: ctx.input.email
);
)
));
client.features.define({
"tag": "user-onboarding",
"name": "User Onboarding",
signals: {
'email-verified': { input: { "code": "string" } },
},
handler: async (ctx) => {
// Step 1: Create user account
user := ctx.step(
'create-user',
async () => {
return ctx.database.insert({
"database": "users-db",
"event": "create-user",
data: {
email: ctx.input.email,
name: ctx.input.name,
password_hash: ctx.input.passwordHash,
"status": "pending_verification",
created_at: new Date().toISOString(),
},
});
},
async (result) => {
// Rollback: Delete user if onboarding fails
ctx.database.delete({
"database": "users-db",
"event": "delete-user",
where: { id: result.id },
});
}
);
// Step 2: Generate verification code
verificationCode := Math.random().toString(36).substring(2, 8).toUpperCase();
ctx.setState('verificationCode', verificationCode);
// Step 3: Send verification email
ctx.step('send-verification', async () => {
ctx.notification.email({
"notification": "transactional",
"event": "email-verification",
recipients: [ctx.input.email],
subject: { "appName": "MyApp" },
template: {
name: ctx.input.name,
code: verificationCode,
"expiresIn": "24 hours",
},
});
});
// Step 4: Wait for email verification
verification := ctx.waitForSignal('email-verified', {
"timeout": "24h",
});
// Validate the code
if (verification.code !== ctx.getState('verificationCode')) {
ctx.triggerRollback('Invalid verification code');
return { "success": false, "reason": "invalid_code" };
}
// Step 5: Activate user account
ctx.step('activate-user', async () => {
ctx.database.update({
"database": "users-db",
"event": "update-user",
where: { id: user.id },
data: {
"status": "active",
verified_at: new Date().toISOString(),
},
});
});
// Step 6: Create user profile
ctx.step('create-profile', async () => {
ctx.database.insert({
"database": "profiles-db",
"event": "create-profile",
data: {
user_id: user.id,
display_name: ctx.input.name,
avatar_url: null,
bio: null,
},
});
});
// Step 7: Send welcome email (non-critical)
ctx.step(
'send-welcome',
async () => {
ctx.notification.email({
"notification": "transactional",
"event": "welcome",
recipients: [ctx.input.email],
template: {
name: ctx.input.name,
"loginUrl": "https://myapp.com/login",
},
});
},
null,
{ "allow_fail": true }
);
return {
"success": true,
userId: user.id,
email: ctx.input.email,
};
},
});
await ductape.features.define({
["tag"] = "user-onboarding",
["name"] = "User Onboarding",
signals: {
'email-verified': { input: { ["code"] = "string" } },
},
handler: async (ctx) => {
// Step 1: Create user account
var user = await ctx.step(
'create-user',
async () => {
return ctx.database.insert({
["database"] = "users-db",
["event"] = "create-user",
data: {
email: ctx.input.email,
name: ctx.input.name,
password_hash: ctx.input.passwordHash,
["status"] = "pending_verification",
created_at: DateTime.UtcNow.toISOString(),
},
});
},
async (result) => {
// Rollback: Delete user if onboarding fails
await ctx.database.delete({
["database"] = "users-db",
["event"] = "delete-user",
where: { id: result.id },
});
}
);
// Step 2: Generate verification code
var verificationCode = Math.random().toString(36).substring(2, 8).toUpperCase();
ctx.setState('verificationCode', verificationCode);
// Step 3: Send verification email
await ctx.step('send-verification', async () => {
await ctx.notification.email({
["notification"] = "transactional",
["event"] = "email-verification",
recipients: [ctx.input.email],
subject: { ["appName"] = "MyApp" },
template: {
name: ctx.input.name,
code: verificationCode,
["expiresIn"] = "24 hours",
},
});
});
// Step 4: Wait for email verification
var verification = await ctx.waitForSignal('email-verified', {
["timeout"] = "24h",
});
// Validate the code
if (verification.code !== ctx.getState('verificationCode')) {
await ctx.triggerRollback('Invalid verification code');
return { ["success"] = false, ["reason"] = "invalid_code" };
}
// Step 5: Activate user account
await ctx.step('activate-user', async () => {
await ctx.database.update({
["database"] = "users-db",
["event"] = "update-user",
where: { id: user.id },
data: {
["status"] = "active",
verified_at: DateTime.UtcNow.toISOString(),
},
});
});
// Step 6: Create user profile
await ctx.step('create-profile', async () => {
await ctx.database.insert({
["database"] = "profiles-db",
["event"] = "create-profile",
data: {
user_id: user.id,
display_name: ctx.input.name,
avatar_url: null,
bio: null,
},
});
});
// Step 7: Send welcome email (non-critical)
await ctx.step(
'send-welcome',
async () => {
await ctx.notification.email({
["notification"] = "transactional",
["event"] = "welcome",
recipients: [ctx.input.email],
template: {
name: ctx.input.name,
["loginUrl"] = "https://myapp.com/login",
},
});
},
null,
{ ["allow_fail"] = true }
);
return {
["success"] = true,
userId: user.id,
email: ctx.input.email,
};
},
});
Document Processing Pipeline
Process uploaded documents with OCR and classification:
- TypeScript
- Java
- Go
- .NET
await ductape.features.define({
tag: 'document-pipeline',
name: 'Document Processing Pipeline',
options: {
timeout: 300000, // 5 minutes for large documents
},
handler: async (ctx) => {
ctx.log.info('Processing document', {
documentId: ctx.input.documentId,
fileName: ctx.input.fileName,
});
// Step 1: Download the document
const document = await ctx.step('download-document', async () => {
return ctx.storage.download({
storage: 'uploads',
event: 'get-document',
input: { file_key: ctx.input.fileKey },
});
});
// Step 2: Run OCR extraction
const ocrResult = await ctx.step(
'extract-text',
async () => {
return ctx.api.run({
app: 'ocr-service',
event: 'extract',
input: {
body: {
content: document.content,
contentType: document.content_type,
options: { language: 'en', detectLayout: true },
},
},
timeout: 60000,
});
},
null,
{ retries: 2 }
);
// Step 3: Classify document type
const classification = await ctx.step('classify-document', async () => {
return ctx.api.run({
app: 'ml-classifier',
event: 'classify',
input: {
body: {
text: ocrResult.text,
categories: ['invoice', 'receipt', 'contract', 'report', 'other'],
},
},
});
});
// Step 4: Extract structured data based on document type
const extractedData = await ctx.step('extract-data', async () => {
const extractionConfig = {
invoice: { fields: ['vendor', 'total', 'date', 'items'] },
receipt: { fields: ['store', 'total', 'date', 'items'] },
contract: { fields: ['parties', 'terms', 'date', 'signatures'] },
report: { fields: ['title', 'author', 'date', 'summary'] },
other: { fields: ['title', 'content'] },
};
return ctx.api.run({
app: 'data-extractor',
event: 'extract',
input: {
body: {
text: ocrResult.text,
documentType: classification.category,
fields: extractionConfig[classification.category],
},
},
});
});
// Step 5: Store processed document
const processedDoc = await ctx.step('store-result', async () => {
return ctx.database.insert({
database: 'documents-db',
event: 'store-processed',
data: {
original_file_key: ctx.input.fileKey,
file_name: ctx.input.fileName,
document_type: classification.category,
confidence: classification.confidence,
extracted_text: ocrResult.text,
extracted_data: extractedData,
processed_at: new Date().toISOString(),
},
});
});
// Step 6: Index for search
await ctx.step(
'index-document',
async () => {
await ctx.api.run({
app: 'search-api',
event: 'index',
input: {
body: {
id: processedDoc.id,
content: ocrResult.text,
metadata: {
type: classification.category,
fileName: ctx.input.fileName,
...extractedData,
},
},
},
});
},
null,
{ allow_fail: true }
);
// Step 7: Notify user
await ctx.step(
'notify-completion',
async () => {
await ctx.notification.push({
notification: 'app-notifications',
event: 'document-processed',
tokens: [ctx.input.deviceToken],
title: { fileName: ctx.input.fileName },
body: { documentType: classification.category },
data: { documentId: processedDoc.id },
});
},
null,
{ allow_fail: true }
);
return {
success: true,
documentId: processedDoc.id,
documentType: classification.category,
confidence: classification.confidence,
};
},
});
ductape.features.define(Map.of(
"tag", "document-pipeline",
"name", "Document Processing Pipeline",
options: Map.of(
"timeout", 300000, // 5 minutes for large documents
),
handler: async (ctx) => Map.of(
ctx.log.info('Processing document', Map.of(
documentId: ctx.input.documentId,
fileName: ctx.input.fileName
));
// Step 1: Download the document
Map<String, Object> document = ctx.step('download-document', async () => Map.of(
return ctx.storage.download(Map.of(
"storage", "uploads",
"event", "get-document",
input: Map.of( file_key: ctx.input.fileKey )
));
));
// Step 2: Run OCR extraction
Map<String, Object> ocrResult = ctx.step(
'extract-text',
async () => Map.of(
return ctx.api.run(Map.of(
"app", "ocr-service",
"event", "extract",
input: Map.of(
body: Map.of(
content: document.content,
contentType: document.content_type,
options: Map.of( "language", "en", "detectLayout", true )
)
),
"timeout", 60000
));
),
null,
Map.of( "retries", 2 )
);
// Step 3: Classify document type
Map<String, Object> classification = ctx.step('classify-document', async () => Map.of(
return ctx.api.run(Map.of(
"app", "ml-classifier",
"event", "classify",
input: Map.of(
body: Map.of(
text: ocrResult.text,
categories: ['invoice', 'receipt', 'contract', 'report', 'other']
)
)
));
));
// Step 4: Extract structured data based on document type
Map<String, Object> extractedData = ctx.step('extract-data', async () => Map.of(
Map<String, Object> extractionConfig = Map.of(
invoice: Map.of( fields: ['vendor', 'total', 'date', 'items'] ),
receipt: Map.of( fields: ['store', 'total', 'date', 'items'] ),
contract: Map.of( fields: ['parties', 'terms', 'date', 'signatures'] ),
report: Map.of( fields: ['title', 'author', 'date', 'summary'] ),
other: Map.of( fields: ['title', 'content'] )
);
return ctx.api.run(Map.of(
"app", "data-extractor",
"event", "extract",
input: Map.of(
body: Map.of(
text: ocrResult.text,
documentType: classification.category,
fields: extractionConfig[classification.category]
)
)
));
));
// Step 5: Store processed document
Map<String, Object> processedDoc = ctx.step('store-result', async () => Map.of(
return ctx.database.insert(Map.of(
"database", "documents-db",
"event", "store-processed",
data: Map.of(
original_file_key: ctx.input.fileKey,
file_name: ctx.input.fileName,
document_type: classification.category,
confidence: classification.confidence,
extracted_text: ocrResult.text,
extracted_data: extractedData,
processed_at: Instant.now().toISOString()
)
));
));
// Step 6: Index for search
ctx.step(
'index-document',
async () => Map.of(
ctx.api.run(Map.of(
"app", "search-api",
"event", "index",
input: Map.of(
body: Map.of(
id: processedDoc.id,
content: ocrResult.text,
metadata: Map.of(
type: classification.category,
fileName: ctx.input.fileName,
...extractedData
)
)
)
));
),
null,
Map.of( "allow_fail", true )
);
// Step 7: Notify user
ctx.step(
'notify-completion',
async () => Map.of(
ctx.notification.push(Map.of(
"notification", "app-notifications",
"event", "document-processed",
tokens: [ctx.input.deviceToken],
title: Map.of( fileName: ctx.input.fileName ),
body: Map.of( documentType: classification.category ),
data: Map.of( documentId: processedDoc.id )
));
),
null,
Map.of( "allow_fail", true )
);
return Map.of(
"success", true,
documentId: processedDoc.id,
documentType: classification.category,
confidence: classification.confidence
);
)
));
client.features.define({
"tag": "document-pipeline",
"name": "Document Processing Pipeline",
options: {
"timeout": 300000, // 5 minutes for large documents
},
handler: async (ctx) => {
ctx.log.info('Processing document', {
documentId: ctx.input.documentId,
fileName: ctx.input.fileName,
});
// Step 1: Download the document
document := ctx.step('download-document', async () => {
return ctx.storage.download({
"storage": "uploads",
"event": "get-document",
input: { file_key: ctx.input.fileKey },
});
});
// Step 2: Run OCR extraction
ocrResult := ctx.step(
'extract-text',
async () => {
return ctx.api.run({
"app": "ocr-service",
"event": "extract",
input: {
body: {
content: document.content,
contentType: document.content_type,
options: { "language": "en", "detectLayout": true },
},
},
"timeout": 60000,
});
},
null,
{ "retries": 2 }
);
// Step 3: Classify document type
classification := ctx.step('classify-document', async () => {
return ctx.api.run({
"app": "ml-classifier",
"event": "classify",
input: {
body: {
text: ocrResult.text,
categories: ['invoice', 'receipt', 'contract', 'report', 'other'],
},
},
});
});
// Step 4: Extract structured data based on document type
extractedData := ctx.step('extract-data', async () => {
extractionConfig := map[string]any{
invoice: { fields: ['vendor', 'total', 'date', 'items'] },
receipt: { fields: ['store', 'total', 'date', 'items'] },
contract: { fields: ['parties', 'terms', 'date', 'signatures'] },
report: { fields: ['title', 'author', 'date', 'summary'] },
other: { fields: ['title', 'content'] },
};
return ctx.api.run({
"app": "data-extractor",
"event": "extract",
input: {
body: {
text: ocrResult.text,
documentType: classification.category,
fields: extractionConfig[classification.category],
},
},
});
});
// Step 5: Store processed document
processedDoc := ctx.step('store-result', async () => {
return ctx.database.insert({
"database": "documents-db",
"event": "store-processed",
data: {
original_file_key: ctx.input.fileKey,
file_name: ctx.input.fileName,
document_type: classification.category,
confidence: classification.confidence,
extracted_text: ocrResult.text,
extracted_data: extractedData,
processed_at: new Date().toISOString(),
},
});
});
// Step 6: Index for search
ctx.step(
'index-document',
async () => {
ctx.api.run({
"app": "search-api",
"event": "index",
input: {
body: {
id: processedDoc.id,
content: ocrResult.text,
metadata: {
type: classification.category,
fileName: ctx.input.fileName,
...extractedData,
},
},
},
});
},
null,
{ "allow_fail": true }
);
// Step 7: Notify user
ctx.step(
'notify-completion',
async () => {
ctx.notification.push({
"notification": "app-notifications",
"event": "document-processed",
tokens: [ctx.input.deviceToken],
title: { fileName: ctx.input.fileName },
body: { documentType: classification.category },
data: { documentId: processedDoc.id },
});
},
null,
{ "allow_fail": true }
);
return {
"success": true,
documentId: processedDoc.id,
documentType: classification.category,
confidence: classification.confidence,
};
},
});
await ductape.features.define({
["tag"] = "document-pipeline",
["name"] = "Document Processing Pipeline",
options: {
["timeout"] = 300000, // 5 minutes for large documents
},
handler: async (ctx) => {
ctx.log.info('Processing document', {
documentId: ctx.input.documentId,
fileName: ctx.input.fileName,
});
// Step 1: Download the document
var document = await ctx.step('download-document', async () => {
return ctx.storage.download({
["storage"] = "uploads",
["event"] = "get-document",
input: { file_key: ctx.input.fileKey },
});
});
// Step 2: Run OCR extraction
var ocrResult = await ctx.step(
'extract-text',
async () => {
return ctx.api.run({
["app"] = "ocr-service",
["event"] = "extract",
input: {
body: {
content: document.content,
contentType: document.content_type,
options: { ["language"] = "en", ["detectLayout"] = true },
},
},
["timeout"] = 60000,
});
},
null,
{ ["retries"] = 2 }
);
// Step 3: Classify document type
var classification = await ctx.step('classify-document', async () => {
return ctx.api.run({
["app"] = "ml-classifier",
["event"] = "classify",
input: {
body: {
text: ocrResult.text,
categories: ['invoice', 'receipt', 'contract', 'report', 'other'],
},
},
});
});
// Step 4: Extract structured data based on document type
var extractedData = await ctx.step('extract-data', async () => {
var extractionConfig = new Dictionary<string, object?>
{
invoice: { fields: ['vendor', 'total', 'date', 'items'] },
receipt: { fields: ['store', 'total', 'date', 'items'] },
contract: { fields: ['parties', 'terms', 'date', 'signatures'] },
report: { fields: ['title', 'author', 'date', 'summary'] },
other: { fields: ['title', 'content'] },
};
return ctx.api.run({
["app"] = "data-extractor",
["event"] = "extract",
input: {
body: {
text: ocrResult.text,
documentType: classification.category,
fields: extractionConfig[classification.category],
},
},
});
});
// Step 5: Store processed document
var processedDoc = await ctx.step('store-result', async () => {
return ctx.database.insert({
["database"] = "documents-db",
["event"] = "store-processed",
data: {
original_file_key: ctx.input.fileKey,
file_name: ctx.input.fileName,
document_type: classification.category,
confidence: classification.confidence,
extracted_text: ocrResult.text,
extracted_data: extractedData,
processed_at: DateTime.UtcNow.toISOString(),
},
});
});
// Step 6: Index for search
await ctx.step(
'index-document',
async () => {
await ctx.api.run({
["app"] = "search-api",
["event"] = "index",
input: {
body: {
id: processedDoc.id,
content: ocrResult.text,
metadata: {
type: classification.category,
fileName: ctx.input.fileName,
...extractedData,
},
},
},
});
},
null,
{ ["allow_fail"] = true }
);
// Step 7: Notify user
await ctx.step(
'notify-completion',
async () => {
await ctx.notification.push({
["notification"] = "app-notifications",
["event"] = "document-processed",
tokens: [ctx.input.deviceToken],
title: { fileName: ctx.input.fileName },
body: { documentType: classification.category },
data: { documentId: processedDoc.id },
});
},
null,
{ ["allow_fail"] = true }
);
return {
["success"] = true,
documentId: processedDoc.id,
documentType: classification.category,
confidence: classification.confidence,
};
},
});
Subscription Billing
Recurring subscription billing with retry logic:
- TypeScript
- Java
- Go
- .NET
await ductape.features.define({
tag: 'subscription-billing',
name: 'Subscription Billing',
options: {
rollback_strategy: 'critical_only',
},
handler: async (ctx) => {
const { subscriptionId, customerId, amount } = ctx.input;
ctx.log.info('Processing subscription billing', { subscriptionId });
// Step 1: Get subscription details
const subscription = await ctx.step('get-subscription', async () => {
return ctx.database.query({
database: 'subscriptions-db',
event: 'get-subscription',
params: { id: subscriptionId },
});
});
if (!subscription || subscription.status !== 'active') {
return { success: false, reason: 'subscription_inactive' };
}
// Step 2: Get payment method
const paymentMethod = await ctx.step('get-payment-method', async () => {
return ctx.api.run({
app: 'stripe',
event: 'get-payment-method',
input: { params: { customerId } },
});
});
if (!paymentMethod.valid) {
// Notify about payment method issue
await ctx.step(
'notify-payment-issue',
async () => {
await ctx.notification.email({
notification: 'billing',
event: 'payment-method-invalid',
recipients: [subscription.email],
template: { updateUrl: 'https://app.com/billing' },
});
},
null,
{ allow_fail: true }
);
return { success: false, reason: 'invalid_payment_method' };
}
// Step 3: Attempt payment (with retries)
let paymentAttempt = 0;
let paymentResult = null;
let lastError = null;
while (paymentAttempt < 3 && !paymentResult) {
paymentAttempt++;
ctx.setState('paymentAttempt', paymentAttempt);
try {
paymentResult = await ctx.step(`payment-attempt-${paymentAttempt}`, async () => {
return ctx.api.run({
app: 'stripe',
event: 'create-charge',
input: {
body: {
amount: amount * 100,
currency: 'usd',
customer: customerId,
payment_method: paymentMethod.id,
metadata: {
subscriptionId,
attempt: paymentAttempt,
},
},
headers: {
'Idempotency-Key': `sub-${subscriptionId}-${ctx.feature_id}-${paymentAttempt}`,
},
},
});
});
} catch (error) {
lastError = error;
ctx.log.warn('Payment attempt failed', {
attempt: paymentAttempt,
error: error.message,
});
if (paymentAttempt < 3) {
// Wait before retry (exponential backoff)
await ctx.sleep(paymentAttempt * 2000);
}
}
}
if (!paymentResult) {
// All payment attempts failed
await ctx.step('mark-payment-failed', async () => {
await ctx.database.update({
database: 'subscriptions-db',
event: 'update-subscription',
where: { id: subscriptionId },
data: {
status: 'past_due',
failed_payments: (subscription.failed_payments || 0) + 1,
last_payment_error: lastError?.message,
},
});
});
await ctx.step(
'notify-payment-failed',
async () => {
await ctx.notification.email({
notification: 'billing',
event: 'payment-failed',
recipients: [subscription.email],
template: {
amount,
attempts: paymentAttempt,
updateUrl: 'https://app.com/billing',
},
});
},
null,
{ allow_fail: true }
);
return {
success: false,
reason: 'payment_failed',
attempts: paymentAttempt,
};
}
// Step 4: Record successful payment (critical)
await ctx.step(
'record-payment',
async () => {
return ctx.database.insert({
database: 'payments-db',
event: 'create-payment',
data: {
subscription_id: subscriptionId,
customer_id: customerId,
charge_id: paymentResult.id,
amount,
status: 'succeeded',
created_at: new Date().toISOString(),
},
});
},
null,
{ critical: true }
);
// Step 5: Update subscription
await ctx.step('update-subscription', async () => {
await ctx.database.update({
database: 'subscriptions-db',
event: 'update-subscription',
where: { id: subscriptionId },
data: {
last_payment_at: new Date().toISOString(),
next_billing_date: new Date(
Date.now() + 30 * 24 * 60 * 60 * 1000
).toISOString(),
failed_payments: 0,
},
});
});
// Step 6: Generate invoice
await ctx.step('generate-invoice', async () => {
const invoice = await ctx.api.run({
app: 'invoice-service',
event: 'generate',
input: {
body: {
subscriptionId,
chargeId: paymentResult.id,
amount,
},
},
});
return invoice;
});
// Step 7: Send receipt (non-critical)
await ctx.step(
'send-receipt',
async () => {
await ctx.notification.email({
notification: 'billing',
event: 'payment-receipt',
recipients: [subscription.email],
template: {
amount,
chargeId: paymentResult.id,
date: new Date().toLocaleDateString(),
},
});
},
null,
{ allow_fail: true }
);
return {
success: true,
chargeId: paymentResult.id,
amount,
nextBillingDate: new Date(
Date.now() + 30 * 24 * 60 * 60 * 1000
).toISOString(),
};
},
});
ductape.features.define(Map.of(
"tag", "subscription-billing",
"name", "Subscription Billing",
options: Map.of(
"rollback_strategy", "critical_only"
),
handler: async (ctx) => Map.of(
Map<String, Object> Map.of( subscriptionId, customerId, amount ) = ctx.input;
ctx.log.info('Processing subscription billing', Map.of( subscriptionId ));
// Step 1: Get subscription details
Map<String, Object> subscription = ctx.step('get-subscription', async () => Map.of(
return ctx.database.query(Map.of(
"database", "subscriptions-db",
"event", "get-subscription",
params: Map.of( id: subscriptionId )
));
));
if (!subscription || subscription.status !== 'active') Map.of(
return Map.of( "success", false, "reason", "subscription_inactive" );
)
// Step 2: Get payment method
Map<String, Object> paymentMethod = ctx.step('get-payment-method', async () => Map.of(
return ctx.api.run(Map.of(
"app", "stripe",
"event", "get-payment-method",
input: Map.of( params: Map.of( customerId ) )
));
));
if (!paymentMethod.valid) Map.of(
// Notify about payment method issue
ctx.step(
'notify-payment-issue',
async () => Map.of(
ctx.notification.email(Map.of(
"notification", "billing",
"event", "payment-method-invalid",
recipients: [subscription.email],
template: Map.of( "updateUrl", "https://app.com/billing" )
));
),
null,
Map.of( "allow_fail", true )
);
return Map.of( "success", false, "reason", "invalid_payment_method" );
)
// Step 3: Attempt payment (with retries)
Map<String, Object> paymentAttempt = 0;
Map<String, Object> paymentResult = null;
Map<String, Object> lastError = null;
while (paymentAttempt < 3 && !paymentResult) Map.of(
paymentAttempt++;
ctx.setState('paymentAttempt', paymentAttempt);
try Map.of(
paymentResult = ctx.step(`payment-attempt-$Map.of(paymentAttempt)`, async () => Map.of(
return ctx.api.run(Map.of(
"app", "stripe",
"event", "create-charge",
input: Map.of(
body: Map.of(
amount: amount * 100,
"currency", "usd",
customer: customerId,
payment_method: paymentMethod.id,
metadata: Map.of(
subscriptionId,
attempt: paymentAttempt
)
),
headers: Map.of(
'Idempotency-Key': `sub-$Map.of(subscriptionId)-$Map.of(ctx.feature_id)-$Map.of(paymentAttempt)`
)
)
));
));
) catch (error) Map.of(
lastError = error;
ctx.log.warn('Payment attempt failed', Map.of(
attempt: paymentAttempt,
error: error.message
));
if (paymentAttempt < 3) Map.of(
// Wait before retry (exponential backoff)
ctx.sleep(paymentAttempt * 2000);
)
)
)
if (!paymentResult) Map.of(
// All payment attempts failed
ctx.step('mark-payment-failed', async () => Map.of(
ctx.database.update(Map.of(
"database", "subscriptions-db",
"event", "update-subscription",
where: Map.of( id: subscriptionId ),
data: Map.of(
"status", "past_due",
failed_payments: (subscription.failed_payments || 0) + 1,
last_payment_error: lastError?.message
)
));
));
ctx.step(
'notify-payment-failed',
async () => Map.of(
ctx.notification.email(Map.of(
"notification", "billing",
"event", "payment-failed",
recipients: [subscription.email],
template: Map.of(
amount,
attempts: paymentAttempt,
"updateUrl", "https://app.com/billing"
)
));
),
null,
Map.of( "allow_fail", true )
);
return Map.of(
"success", false,
"reason", "payment_failed",
attempts: paymentAttempt
);
)
// Step 4: Record successful payment (critical)
ctx.step(
'record-payment',
async () => Map.of(
return ctx.database.insert(Map.of(
"database", "payments-db",
"event", "create-payment",
data: Map.of(
subscription_id: subscriptionId,
customer_id: customerId,
charge_id: paymentResult.id,
amount,
"status", "succeeded",
created_at: Instant.now().toISOString()
)
));
),
null,
Map.of( "critical", true )
);
// Step 5: Update subscription
ctx.step('update-subscription', async () => Map.of(
ctx.database.update(Map.of(
"database", "subscriptions-db",
"event", "update-subscription",
where: Map.of( id: subscriptionId ),
data: Map.of(
last_payment_at: Instant.now().toISOString(),
next_billing_date: new Date(
Date.now() + 30 * 24 * 60 * 60 * 1000
).toISOString(),
"failed_payments", 0
)
));
));
// Step 6: Generate invoice
ctx.step('generate-invoice', async () => Map.of(
Map<String, Object> invoice = ctx.api.run(Map.of(
"app", "invoice-service",
"event", "generate",
input: Map.of(
body: Map.of(
subscriptionId,
chargeId: paymentResult.id,
amount
)
)
));
return invoice;
));
// Step 7: Send receipt (non-critical)
ctx.step(
'send-receipt',
async () => Map.of(
ctx.notification.email(Map.of(
"notification", "billing",
"event", "payment-receipt",
recipients: [subscription.email],
template: Map.of(
amount,
chargeId: paymentResult.id,
date: Instant.now().toLocaleDateString()
)
));
),
null,
Map.of( "allow_fail", true )
);
return Map.of(
"success", true,
chargeId: paymentResult.id,
amount,
nextBillingDate: new Date(
Date.now() + 30 * 24 * 60 * 60 * 1000
).toISOString()
);
)
));
client.features.define({
"tag": "subscription-billing",
"name": "Subscription Billing",
options: {
"rollback_strategy": "critical_only",
},
handler: async (ctx) => {
const { subscriptionId, customerId, amount } = ctx.input;
ctx.log.info('Processing subscription billing', { subscriptionId });
// Step 1: Get subscription details
subscription := ctx.step('get-subscription', async () => {
return ctx.database.query({
"database": "subscriptions-db",
"event": "get-subscription",
params: { id: subscriptionId },
});
});
if (!subscription || subscription.status !== 'active') {
return { "success": false, "reason": "subscription_inactive" };
}
// Step 2: Get payment method
paymentMethod := ctx.step('get-payment-method', async () => {
return ctx.api.run({
"app": "stripe",
"event": "get-payment-method",
input: { params: { customerId } },
});
});
if (!paymentMethod.valid) {
// Notify about payment method issue
ctx.step(
'notify-payment-issue',
async () => {
ctx.notification.email({
"notification": "billing",
"event": "payment-method-invalid",
recipients: [subscription.email],
template: { "updateUrl": "https://app.com/billing" },
});
},
null,
{ "allow_fail": true }
);
return { "success": false, "reason": "invalid_payment_method" };
}
// Step 3: Attempt payment (with retries)
paymentAttempt := 0;
paymentResult := null;
lastError := null;
while (paymentAttempt < 3 && !paymentResult) {
paymentAttempt++;
ctx.setState('paymentAttempt', paymentAttempt);
try {
paymentResult = ctx.step(`payment-attempt-${paymentAttempt}`, async () => {
return ctx.api.run({
"app": "stripe",
"event": "create-charge",
input: {
body: {
amount: amount * 100,
"currency": "usd",
customer: customerId,
payment_method: paymentMethod.id,
metadata: {
subscriptionId,
attempt: paymentAttempt,
},
},
headers: {
'Idempotency-Key': `sub-${subscriptionId}-${ctx.feature_id}-${paymentAttempt}`,
},
},
});
});
} catch (error) {
lastError = error;
ctx.log.warn('Payment attempt failed', {
attempt: paymentAttempt,
error: error.message,
});
if (paymentAttempt < 3) {
// Wait before retry (exponential backoff)
ctx.sleep(paymentAttempt * 2000);
}
}
}
if (!paymentResult) {
// All payment attempts failed
ctx.step('mark-payment-failed', async () => {
ctx.database.update({
"database": "subscriptions-db",
"event": "update-subscription",
where: { id: subscriptionId },
data: {
"status": "past_due",
failed_payments: (subscription.failed_payments || 0) + 1,
last_payment_error: lastError?.message,
},
});
});
ctx.step(
'notify-payment-failed',
async () => {
ctx.notification.email({
"notification": "billing",
"event": "payment-failed",
recipients: [subscription.email],
template: {
amount,
attempts: paymentAttempt,
"updateUrl": "https://app.com/billing",
},
});
},
null,
{ "allow_fail": true }
);
return {
"success": false,
"reason": "payment_failed",
attempts: paymentAttempt,
};
}
// Step 4: Record successful payment (critical)
ctx.step(
'record-payment',
async () => {
return ctx.database.insert({
"database": "payments-db",
"event": "create-payment",
data: {
subscription_id: subscriptionId,
customer_id: customerId,
charge_id: paymentResult.id,
amount,
"status": "succeeded",
created_at: new Date().toISOString(),
},
});
},
null,
{ "critical": true }
);
// Step 5: Update subscription
ctx.step('update-subscription', async () => {
ctx.database.update({
"database": "subscriptions-db",
"event": "update-subscription",
where: { id: subscriptionId },
data: {
last_payment_at: new Date().toISOString(),
next_billing_date: new Date(
Date.now() + 30 * 24 * 60 * 60 * 1000
).toISOString(),
"failed_payments": 0,
},
});
});
// Step 6: Generate invoice
ctx.step('generate-invoice', async () => {
invoice := ctx.api.run({
"app": "invoice-service",
"event": "generate",
input: {
body: {
subscriptionId,
chargeId: paymentResult.id,
amount,
},
},
});
return invoice;
});
// Step 7: Send receipt (non-critical)
ctx.step(
'send-receipt',
async () => {
ctx.notification.email({
"notification": "billing",
"event": "payment-receipt",
recipients: [subscription.email],
template: {
amount,
chargeId: paymentResult.id,
date: new Date().toLocaleDateString(),
},
});
},
null,
{ "allow_fail": true }
);
return {
"success": true,
chargeId: paymentResult.id,
amount,
nextBillingDate: new Date(
Date.now() + 30 * 24 * 60 * 60 * 1000
).toISOString(),
};
},
});
await ductape.features.define({
["tag"] = "subscription-billing",
["name"] = "Subscription Billing",
options: {
["rollback_strategy"] = "critical_only",
},
handler: async (ctx) => {
var { subscriptionId, customerId, amount } = ctx.input;
ctx.log.info('Processing subscription billing', { subscriptionId });
// Step 1: Get subscription details
var subscription = await ctx.step('get-subscription', async () => {
return ctx.database.query({
["database"] = "subscriptions-db",
["event"] = "get-subscription",
params: { id: subscriptionId },
});
});
if (!subscription || subscription.status !== 'active') {
return { ["success"] = false, ["reason"] = "subscription_inactive" };
}
// Step 2: Get payment method
var paymentMethod = await ctx.step('get-payment-method', async () => {
return ctx.api.run({
["app"] = "stripe",
["event"] = "get-payment-method",
input: { params: { customerId } },
});
});
if (!paymentMethod.valid) {
// Notify about payment method issue
await ctx.step(
'notify-payment-issue',
async () => {
await ctx.notification.email({
["notification"] = "billing",
["event"] = "payment-method-invalid",
recipients: [subscription.email],
template: { ["updateUrl"] = "https://app.com/billing" },
});
},
null,
{ ["allow_fail"] = true }
);
return { ["success"] = false, ["reason"] = "invalid_payment_method" };
}
// Step 3: Attempt payment (with retries)
var paymentAttempt = 0;
var paymentResult = null;
var lastError = null;
while (paymentAttempt < 3 && !paymentResult) {
paymentAttempt++;
ctx.setState('paymentAttempt', paymentAttempt);
try {
paymentResult = await ctx.step(`payment-attempt-${paymentAttempt}`, async () => {
return ctx.api.run({
["app"] = "stripe",
["event"] = "create-charge",
input: {
body: {
amount: amount * 100,
["currency"] = "usd",
customer: customerId,
payment_method: paymentMethod.id,
metadata: {
subscriptionId,
attempt: paymentAttempt,
},
},
headers: {
'Idempotency-Key': `sub-${subscriptionId}-${ctx.feature_id}-${paymentAttempt}`,
},
},
});
});
} catch (error) {
lastError = error;
ctx.log.warn('Payment attempt failed', {
attempt: paymentAttempt,
error: error.message,
});
if (paymentAttempt < 3) {
// Wait before retry (exponential backoff)
await ctx.sleep(paymentAttempt * 2000);
}
}
}
if (!paymentResult) {
// All payment attempts failed
await ctx.step('mark-payment-failed', async () => {
await ctx.database.update({
["database"] = "subscriptions-db",
["event"] = "update-subscription",
where: { id: subscriptionId },
data: {
["status"] = "past_due",
failed_payments: (subscription.failed_payments || 0) + 1,
last_payment_error: lastError?.message,
},
});
});
await ctx.step(
'notify-payment-failed',
async () => {
await ctx.notification.email({
["notification"] = "billing",
["event"] = "payment-failed",
recipients: [subscription.email],
template: {
amount,
attempts: paymentAttempt,
["updateUrl"] = "https://app.com/billing",
},
});
},
null,
{ ["allow_fail"] = true }
);
return {
["success"] = false,
["reason"] = "payment_failed",
attempts: paymentAttempt,
};
}
// Step 4: Record successful payment (critical)
await ctx.step(
'record-payment',
async () => {
return ctx.database.insert({
["database"] = "payments-db",
["event"] = "create-payment",
data: {
subscription_id: subscriptionId,
customer_id: customerId,
charge_id: paymentResult.id,
amount,
["status"] = "succeeded",
created_at: DateTime.UtcNow.toISOString(),
},
});
},
null,
{ ["critical"] = true }
);
// Step 5: Update subscription
await ctx.step('update-subscription', async () => {
await ctx.database.update({
["database"] = "subscriptions-db",
["event"] = "update-subscription",
where: { id: subscriptionId },
data: {
last_payment_at: DateTime.UtcNow.toISOString(),
next_billing_date: new Date(
Date.now() + 30 * 24 * 60 * 60 * 1000
).toISOString(),
["failed_payments"] = 0,
},
});
});
// Step 6: Generate invoice
await ctx.step('generate-invoice', async () => {
var invoice = await ctx.api.run({
["app"] = "invoice-service",
["event"] = "generate",
input: {
body: {
subscriptionId,
chargeId: paymentResult.id,
amount,
},
},
});
return invoice;
});
// Step 7: Send receipt (non-critical)
await ctx.step(
'send-receipt',
async () => {
await ctx.notification.email({
["notification"] = "billing",
["event"] = "payment-receipt",
recipients: [subscription.email],
template: {
amount,
chargeId: paymentResult.id,
date: DateTime.UtcNow.toLocaleDateString(),
},
});
},
null,
{ ["allow_fail"] = true }
);
return {
["success"] = true,
chargeId: paymentResult.id,
amount,
nextBillingDate: new Date(
Date.now() + 30 * 24 * 60 * 60 * 1000
).toISOString(),
};
},
});
Approval Feature
Multi-level approval process with escalation:
- TypeScript
- Java
- Go
- .NET
await ductape.features.define({
tag: 'expense-approval',
name: 'Expense Approval',
signals: {
'manager-decision': { input: { approved: 'boolean', comments: 'string' } },
'director-decision': { input: { approved: 'boolean', comments: 'string' } },
'finance-decision': { input: { approved: 'boolean', comments: 'string' } },
},
queries: {
'getStatus': {
handler: (ctx) => ({
requestId: ctx.input.requestId,
amount: ctx.input.amount,
status: ctx.getState('status') || 'pending',
currentApprover: ctx.getState('currentApprover'),
approvals: ctx.getState('approvals') || [],
}),
},
},
handler: async (ctx) => {
const { requestId, employeeId, amount, description } = ctx.input;
ctx.setState('status', 'pending');
ctx.setState('approvals', []);
// Step 1: Create expense request
const request = await ctx.step('create-request', async () => {
return ctx.database.insert({
database: 'expenses-db',
event: 'create-request',
data: {
request_id: requestId,
employee_id: employeeId,
amount,
description,
status: 'pending_approval',
created_at: new Date().toISOString(),
},
});
});
// Step 2: Get approval chain
const approvalChain = await ctx.step('get-approval-chain', async () => {
// Determine required approvers based on amount
const chain = [];
chain.push({ level: 'manager', required: true });
if (amount > 1000) {
chain.push({ level: 'director', required: true });
}
if (amount > 5000) {
chain.push({ level: 'finance', required: true });
}
return chain;
});
ctx.setState('approvalChain', approvalChain);
// Step 3: Manager approval
ctx.setState('currentApprover', 'manager');
await ctx.step('notify-manager', async () => {
await ctx.notification.email({
notification: 'approvals',
event: 'approval-request',
recipients: [ctx.input.managerEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
const managerDecision = await ctx.waitForSignal('manager-decision', {
timeout: '48h',
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
level: 'manager',
approved: managerDecision.approved,
comments: managerDecision.comments,
timestamp: Date.now(),
},
]);
if (!managerDecision.approved) {
await ctx.step('mark-rejected', async () => {
await ctx.database.update({
database: 'expenses-db',
event: 'update-request',
where: { request_id: requestId },
data: {
status: 'rejected',
rejected_by: 'manager',
rejection_reason: managerDecision.comments,
},
});
});
await ctx.step('notify-rejection', async () => {
await ctx.notification.email({
notification: 'approvals',
event: 'request-rejected',
recipients: [ctx.input.employeeEmail],
template: {
amount,
rejectedBy: 'Manager',
reason: managerDecision.comments,
},
});
});
return {
success: false,
status: 'rejected',
rejectedBy: 'manager',
reason: managerDecision.comments,
};
}
// Step 4: Director approval (if needed)
if (amount > 1000) {
ctx.setState('currentApprover', 'director');
await ctx.step('notify-director', async () => {
await ctx.notification.email({
notification: 'approvals',
event: 'approval-request',
recipients: [ctx.input.directorEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
managerApproved: true,
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
const directorDecision = await ctx.waitForSignal('director-decision', {
timeout: '48h',
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
level: 'director',
approved: directorDecision.approved,
comments: directorDecision.comments,
timestamp: Date.now(),
},
]);
if (!directorDecision.approved) {
// Similar rejection handling...
return { success: false, status: 'rejected', rejectedBy: 'director' };
}
}
// Step 5: Finance approval (if needed)
if (amount > 5000) {
ctx.setState('currentApprover', 'finance');
await ctx.step('notify-finance', async () => {
await ctx.notification.email({
notification: 'approvals',
event: 'approval-request',
recipients: [ctx.input.financeEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
previousApprovals: ctx.getState('approvals'),
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
const financeDecision = await ctx.waitForSignal('finance-decision', {
timeout: '72h',
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
level: 'finance',
approved: financeDecision.approved,
comments: financeDecision.comments,
timestamp: Date.now(),
},
]);
if (!financeDecision.approved) {
return { success: false, status: 'rejected', rejectedBy: 'finance' };
}
}
// Step 6: Mark as approved
ctx.setState('status', 'approved');
await ctx.step('mark-approved', async () => {
await ctx.database.update({
database: 'expenses-db',
event: 'update-request',
where: { request_id: requestId },
data: {
status: 'approved',
approved_at: new Date().toISOString(),
approvals: ctx.getState('approvals'),
},
});
});
// Step 7: Initiate reimbursement
await ctx.step('initiate-reimbursement', async () => {
await ctx.messaging.produce({
event: 'payroll-events:process-reimbursement',
message: {
requestId,
employeeId,
amount,
approvals: ctx.getState('approvals'),
},
});
});
// Step 8: Notify employee
await ctx.step('notify-approval', async () => {
await ctx.notification.email({
notification: 'approvals',
event: 'request-approved',
recipients: [ctx.input.employeeEmail],
template: {
amount,
description,
expectedPayment: 'Next payroll cycle',
},
});
});
return {
success: true,
status: 'approved',
approvals: ctx.getState('approvals'),
};
},
});
ductape.features.define(Map.of(
"tag", "expense-approval",
"name", "Expense Approval",
signals: Map.of(
'manager-decision': Map.of( input: Map.of( "approved", "boolean", "comments", "string" ) ),
'director-decision': Map.of( input: Map.of( "approved", "boolean", "comments", "string" ) ),
'finance-decision': Map.of( input: Map.of( "approved", "boolean", "comments", "string" ) )
),
queries: Map.of(
'getStatus': Map.of(
handler: (ctx) => (Map.of(
requestId: ctx.input.requestId,
amount: ctx.input.amount,
status: ctx.getState('status') || 'pending',
currentApprover: ctx.getState('currentApprover'),
approvals: ctx.getState('approvals') || []
))
)
),
handler: async (ctx) => Map.of(
Map<String, Object> Map.of( requestId, employeeId, amount, description ) = ctx.input;
ctx.setState('status', 'pending');
ctx.setState('approvals', []);
// Step 1: Create expense request
Map<String, Object> request = ctx.step('create-request', async () => Map.of(
return ctx.database.insert(Map.of(
"database", "expenses-db",
"event", "create-request",
data: Map.of(
request_id: requestId,
employee_id: employeeId,
amount,
description,
"status", "pending_approval",
created_at: Instant.now().toISOString()
)
));
));
// Step 2: Get approval chain
Map<String, Object> approvalChain = ctx.step('get-approval-chain', async () => Map.of(
// Determine required approvers based on amount
Map<String, Object> chain = [];
chain.push(Map.of( "level", "manager", "required", true ));
if (amount > 1000) Map.of(
chain.push(Map.of( "level", "director", "required", true ));
)
if (amount > 5000) Map.of(
chain.push(Map.of( "level", "finance", "required", true ));
)
return chain;
));
ctx.setState('approvalChain', approvalChain);
// Step 3: Manager approval
ctx.setState('currentApprover', 'manager');
ctx.step('notify-manager', async () => Map.of(
ctx.notification.email(Map.of(
"notification", "approvals",
"event", "approval-request",
recipients: [ctx.input.managerEmail],
template: Map.of(
employeeName: ctx.input.employeeName,
amount,
description,
approveUrl: `https://app.com/approve/$Map.of(requestId)`
)
));
));
Map<String, Object> managerDecision = ctx.waitForSignal('manager-decision', Map.of(
"timeout", "48h"
));
ctx.setState('approvals', [
...ctx.getState('approvals'),
Map.of(
"level", "manager",
approved: managerDecision.approved,
comments: managerDecision.comments,
timestamp: Date.now()
),
]);
if (!managerDecision.approved) Map.of(
ctx.step('mark-rejected', async () => Map.of(
ctx.database.update(Map.of(
"database", "expenses-db",
"event", "update-request",
where: Map.of( request_id: requestId ),
data: Map.of(
"status", "rejected",
"rejected_by", "manager",
rejection_reason: managerDecision.comments
)
));
));
ctx.step('notify-rejection', async () => Map.of(
ctx.notification.email(Map.of(
"notification", "approvals",
"event", "request-rejected",
recipients: [ctx.input.employeeEmail],
template: Map.of(
amount,
"rejectedBy", "Manager",
reason: managerDecision.comments
)
));
));
return Map.of(
"success", false,
"status", "rejected",
"rejectedBy", "manager",
reason: managerDecision.comments
);
)
// Step 4: Director approval (if needed)
if (amount > 1000) Map.of(
ctx.setState('currentApprover', 'director');
ctx.step('notify-director', async () => Map.of(
ctx.notification.email(Map.of(
"notification", "approvals",
"event", "approval-request",
recipients: [ctx.input.directorEmail],
template: Map.of(
employeeName: ctx.input.employeeName,
amount,
description,
"managerApproved", true,
approveUrl: `https://app.com/approve/$Map.of(requestId)`
)
));
));
Map<String, Object> directorDecision = ctx.waitForSignal('director-decision', Map.of(
"timeout", "48h"
));
ctx.setState('approvals', [
...ctx.getState('approvals'),
Map.of(
"level", "director",
approved: directorDecision.approved,
comments: directorDecision.comments,
timestamp: Date.now()
),
]);
if (!directorDecision.approved) Map.of(
// Similar rejection handling...
return Map.of( "success", false, "status", "rejected", "rejectedBy", "director" );
)
)
// Step 5: Finance approval (if needed)
if (amount > 5000) Map.of(
ctx.setState('currentApprover', 'finance');
ctx.step('notify-finance', async () => Map.of(
ctx.notification.email(Map.of(
"notification", "approvals",
"event", "approval-request",
recipients: [ctx.input.financeEmail],
template: Map.of(
employeeName: ctx.input.employeeName,
amount,
description,
previousApprovals: ctx.getState('approvals'),
approveUrl: `https://app.com/approve/$Map.of(requestId)`
)
));
));
Map<String, Object> financeDecision = ctx.waitForSignal('finance-decision', Map.of(
"timeout", "72h"
));
ctx.setState('approvals', [
...ctx.getState('approvals'),
Map.of(
"level", "finance",
approved: financeDecision.approved,
comments: financeDecision.comments,
timestamp: Date.now()
),
]);
if (!financeDecision.approved) Map.of(
return Map.of( "success", false, "status", "rejected", "rejectedBy", "finance" );
)
)
// Step 6: Mark as approved
ctx.setState('status', 'approved');
ctx.step('mark-approved', async () => Map.of(
ctx.database.update(Map.of(
"database", "expenses-db",
"event", "update-request",
where: Map.of( request_id: requestId ),
data: Map.of(
"status", "approved",
approved_at: Instant.now().toISOString(),
approvals: ctx.getState('approvals')
)
));
));
// Step 7: Initiate reimbursement
ctx.step('initiate-reimbursement', async () => Map.of(
ctx.messaging.produce(Map.of(
"event", "payroll-events:process-reimbursement",
message: Map.of(
requestId,
employeeId,
amount,
approvals: ctx.getState('approvals')
)
));
));
// Step 8: Notify employee
ctx.step('notify-approval', async () => Map.of(
ctx.notification.email(Map.of(
"notification", "approvals",
"event", "request-approved",
recipients: [ctx.input.employeeEmail],
template: Map.of(
amount,
description,
"expectedPayment", "Next payroll cycle"
)
));
));
return Map.of(
"success", true,
"status", "approved",
approvals: ctx.getState('approvals')
);
)
));
client.features.define({
"tag": "expense-approval",
"name": "Expense Approval",
signals: {
'manager-decision': { input: { "approved": "boolean", "comments": "string" } },
'director-decision': { input: { "approved": "boolean", "comments": "string" } },
'finance-decision': { input: { "approved": "boolean", "comments": "string" } },
},
queries: {
'getStatus': {
handler: (ctx) => ({
requestId: ctx.input.requestId,
amount: ctx.input.amount,
status: ctx.getState('status') || 'pending',
currentApprover: ctx.getState('currentApprover'),
approvals: ctx.getState('approvals') || [],
}),
},
},
handler: async (ctx) => {
const { requestId, employeeId, amount, description } = ctx.input;
ctx.setState('status', 'pending');
ctx.setState('approvals', []);
// Step 1: Create expense request
request := ctx.step('create-request', async () => {
return ctx.database.insert({
"database": "expenses-db",
"event": "create-request",
data: {
request_id: requestId,
employee_id: employeeId,
amount,
description,
"status": "pending_approval",
created_at: new Date().toISOString(),
},
});
});
// Step 2: Get approval chain
approvalChain := ctx.step('get-approval-chain', async () => {
// Determine required approvers based on amount
chain := [];
chain.push({ "level": "manager", "required": true });
if (amount > 1000) {
chain.push({ "level": "director", "required": true });
}
if (amount > 5000) {
chain.push({ "level": "finance", "required": true });
}
return chain;
});
ctx.setState('approvalChain', approvalChain);
// Step 3: Manager approval
ctx.setState('currentApprover', 'manager');
ctx.step('notify-manager', async () => {
ctx.notification.email({
"notification": "approvals",
"event": "approval-request",
recipients: [ctx.input.managerEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
managerDecision := ctx.waitForSignal('manager-decision', {
"timeout": "48h",
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
"level": "manager",
approved: managerDecision.approved,
comments: managerDecision.comments,
timestamp: Date.now(),
},
]);
if (!managerDecision.approved) {
ctx.step('mark-rejected', async () => {
ctx.database.update({
"database": "expenses-db",
"event": "update-request",
where: { request_id: requestId },
data: {
"status": "rejected",
"rejected_by": "manager",
rejection_reason: managerDecision.comments,
},
});
});
ctx.step('notify-rejection', async () => {
ctx.notification.email({
"notification": "approvals",
"event": "request-rejected",
recipients: [ctx.input.employeeEmail],
template: {
amount,
"rejectedBy": "Manager",
reason: managerDecision.comments,
},
});
});
return {
"success": false,
"status": "rejected",
"rejectedBy": "manager",
reason: managerDecision.comments,
};
}
// Step 4: Director approval (if needed)
if (amount > 1000) {
ctx.setState('currentApprover', 'director');
ctx.step('notify-director', async () => {
ctx.notification.email({
"notification": "approvals",
"event": "approval-request",
recipients: [ctx.input.directorEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
"managerApproved": true,
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
directorDecision := ctx.waitForSignal('director-decision', {
"timeout": "48h",
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
"level": "director",
approved: directorDecision.approved,
comments: directorDecision.comments,
timestamp: Date.now(),
},
]);
if (!directorDecision.approved) {
// Similar rejection handling...
return { "success": false, "status": "rejected", "rejectedBy": "director" };
}
}
// Step 5: Finance approval (if needed)
if (amount > 5000) {
ctx.setState('currentApprover', 'finance');
ctx.step('notify-finance', async () => {
ctx.notification.email({
"notification": "approvals",
"event": "approval-request",
recipients: [ctx.input.financeEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
previousApprovals: ctx.getState('approvals'),
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
financeDecision := ctx.waitForSignal('finance-decision', {
"timeout": "72h",
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
"level": "finance",
approved: financeDecision.approved,
comments: financeDecision.comments,
timestamp: Date.now(),
},
]);
if (!financeDecision.approved) {
return { "success": false, "status": "rejected", "rejectedBy": "finance" };
}
}
// Step 6: Mark as approved
ctx.setState('status', 'approved');
ctx.step('mark-approved', async () => {
ctx.database.update({
"database": "expenses-db",
"event": "update-request",
where: { request_id: requestId },
data: {
"status": "approved",
approved_at: new Date().toISOString(),
approvals: ctx.getState('approvals'),
},
});
});
// Step 7: Initiate reimbursement
ctx.step('initiate-reimbursement', async () => {
ctx.messaging.produce({
"event": "payroll-events:process-reimbursement",
message: {
requestId,
employeeId,
amount,
approvals: ctx.getState('approvals'),
},
});
});
// Step 8: Notify employee
ctx.step('notify-approval', async () => {
ctx.notification.email({
"notification": "approvals",
"event": "request-approved",
recipients: [ctx.input.employeeEmail],
template: {
amount,
description,
"expectedPayment": "Next payroll cycle",
},
});
});
return {
"success": true,
"status": "approved",
approvals: ctx.getState('approvals'),
};
},
});
await ductape.features.define({
["tag"] = "expense-approval",
["name"] = "Expense Approval",
signals: {
'manager-decision': { input: { ["approved"] = "boolean", ["comments"] = "string" } },
'director-decision': { input: { ["approved"] = "boolean", ["comments"] = "string" } },
'finance-decision': { input: { ["approved"] = "boolean", ["comments"] = "string" } },
},
queries: {
'getStatus': {
handler: (ctx) => ({
requestId: ctx.input.requestId,
amount: ctx.input.amount,
status: ctx.getState('status') || 'pending',
currentApprover: ctx.getState('currentApprover'),
approvals: ctx.getState('approvals') || [],
}),
},
},
handler: async (ctx) => {
var { requestId, employeeId, amount, description } = ctx.input;
ctx.setState('status', 'pending');
ctx.setState('approvals', []);
// Step 1: Create expense request
var request = await ctx.step('create-request', async () => {
return ctx.database.insert({
["database"] = "expenses-db",
["event"] = "create-request",
data: {
request_id: requestId,
employee_id: employeeId,
amount,
description,
["status"] = "pending_approval",
created_at: DateTime.UtcNow.toISOString(),
},
});
});
// Step 2: Get approval chain
var approvalChain = await ctx.step('get-approval-chain', async () => {
// Determine required approvers based on amount
var chain = [];
chain.push({ ["level"] = "manager", ["required"] = true });
if (amount > 1000) {
chain.push({ ["level"] = "director", ["required"] = true });
}
if (amount > 5000) {
chain.push({ ["level"] = "finance", ["required"] = true });
}
return chain;
});
ctx.setState('approvalChain', approvalChain);
// Step 3: Manager approval
ctx.setState('currentApprover', 'manager');
await ctx.step('notify-manager', async () => {
await ctx.notification.email({
["notification"] = "approvals",
["event"] = "approval-request",
recipients: [ctx.input.managerEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
var managerDecision = await ctx.waitForSignal('manager-decision', {
["timeout"] = "48h",
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
["level"] = "manager",
approved: managerDecision.approved,
comments: managerDecision.comments,
timestamp: Date.now(),
},
]);
if (!managerDecision.approved) {
await ctx.step('mark-rejected', async () => {
await ctx.database.update({
["database"] = "expenses-db",
["event"] = "update-request",
where: { request_id: requestId },
data: {
["status"] = "rejected",
["rejected_by"] = "manager",
rejection_reason: managerDecision.comments,
},
});
});
await ctx.step('notify-rejection', async () => {
await ctx.notification.email({
["notification"] = "approvals",
["event"] = "request-rejected",
recipients: [ctx.input.employeeEmail],
template: {
amount,
["rejectedBy"] = "Manager",
reason: managerDecision.comments,
},
});
});
return {
["success"] = false,
["status"] = "rejected",
["rejectedBy"] = "manager",
reason: managerDecision.comments,
};
}
// Step 4: Director approval (if needed)
if (amount > 1000) {
ctx.setState('currentApprover', 'director');
await ctx.step('notify-director', async () => {
await ctx.notification.email({
["notification"] = "approvals",
["event"] = "approval-request",
recipients: [ctx.input.directorEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
["managerApproved"] = true,
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
var directorDecision = await ctx.waitForSignal('director-decision', {
["timeout"] = "48h",
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
["level"] = "director",
approved: directorDecision.approved,
comments: directorDecision.comments,
timestamp: Date.now(),
},
]);
if (!directorDecision.approved) {
// Similar rejection handling...
return { ["success"] = false, ["status"] = "rejected", ["rejectedBy"] = "director" };
}
}
// Step 5: Finance approval (if needed)
if (amount > 5000) {
ctx.setState('currentApprover', 'finance');
await ctx.step('notify-finance', async () => {
await ctx.notification.email({
["notification"] = "approvals",
["event"] = "approval-request",
recipients: [ctx.input.financeEmail],
template: {
employeeName: ctx.input.employeeName,
amount,
description,
previousApprovals: ctx.getState('approvals'),
approveUrl: `https://app.com/approve/${requestId}`,
},
});
});
var financeDecision = await ctx.waitForSignal('finance-decision', {
["timeout"] = "72h",
});
ctx.setState('approvals', [
...ctx.getState('approvals'),
{
["level"] = "finance",
approved: financeDecision.approved,
comments: financeDecision.comments,
timestamp: Date.now(),
},
]);
if (!financeDecision.approved) {
return { ["success"] = false, ["status"] = "rejected", ["rejectedBy"] = "finance" };
}
}
// Step 6: Mark as approved
ctx.setState('status', 'approved');
await ctx.step('mark-approved', async () => {
await ctx.database.update({
["database"] = "expenses-db",
["event"] = "update-request",
where: { request_id: requestId },
data: {
["status"] = "approved",
approved_at: DateTime.UtcNow.toISOString(),
approvals: ctx.getState('approvals'),
},
});
});
// Step 7: Initiate reimbursement
await ctx.step('initiate-reimbursement', async () => {
await ctx.messaging.produce({
["event"] = "payroll-events:process-reimbursement",
message: {
requestId,
employeeId,
amount,
approvals: ctx.getState('approvals'),
},
});
});
// Step 8: Notify employee
await ctx.step('notify-approval', async () => {
await ctx.notification.email({
["notification"] = "approvals",
["event"] = "request-approved",
recipients: [ctx.input.employeeEmail],
template: {
amount,
description,
["expectedPayment"] = "Next payroll cycle",
},
});
});
return {
["success"] = true,
["status"] = "approved",
approvals: ctx.getState('approvals'),
};
},
});
Data Sync Feature
Synchronize data between systems with conflict resolution:
- TypeScript
- Java
- Go
- .NET
await ductape.features.define({
tag: 'data-sync',
name: 'Data Synchronization',
options: {
timeout: 600000,
rollback_strategy: 'to_checkpoint',
},
handler: async (ctx) => {
const { sourceSystem, targetSystem, entityType, batchSize = 100 } = ctx.input;
ctx.log.info('Starting data sync', { sourceSystem, targetSystem, entityType });
// Step 1: Get last sync checkpoint
const lastSync = await ctx.step('get-last-sync', async () => {
const result = await ctx.database.query({
database: 'sync-db',
event: 'get-sync-state',
params: { source: sourceSystem, target: targetSystem, entity: entityType },
});
return result[0] || { lastSyncedAt: null, lastSyncedId: null };
});
// Step 2: Fetch changed records from source
const sourceRecords = await ctx.step('fetch-source-records', async () => {
return ctx.api.run({
app: sourceSystem,
event: 'get-records',
input: {
body: {
entityType,
modifiedAfter: lastSync.lastSyncedAt,
limit: batchSize,
},
},
});
});
if (sourceRecords.records.length === 0) {
ctx.log.info('No records to sync');
return { success: true, synced: 0, conflicts: 0 };
}
ctx.setState('totalRecords', sourceRecords.records.length);
ctx.setState('syncedRecords', 0);
ctx.setState('conflicts', []);
await ctx.checkpoint('records-fetched', {
count: sourceRecords.records.length,
});
// Step 3: Process each record
for (const record of sourceRecords.records) {
await ctx.step(`sync-record-${record.id}`, async () => {
// Check for existing record in target
const existing = await ctx.api.run({
app: targetSystem,
event: 'get-record',
input: { params: { id: record.id } },
});
if (existing && existing.modifiedAt > record.modifiedAt) {
// Conflict: target is newer
const conflicts = ctx.getState('conflicts') || [];
conflicts.push({
recordId: record.id,
sourceModified: record.modifiedAt,
targetModified: existing.modifiedAt,
});
ctx.setState('conflicts', conflicts);
ctx.log.warn('Sync conflict detected', { recordId: record.id });
return { skipped: true, reason: 'conflict' };
}
// Sync the record
if (existing) {
await ctx.api.run({
app: targetSystem,
event: 'update-record',
input: { body: record },
});
} else {
await ctx.api.run({
app: targetSystem,
event: 'create-record',
input: { body: record },
});
}
const synced = (ctx.getState('syncedRecords') || 0) + 1;
ctx.setState('syncedRecords', synced);
return { synced: true };
});
}
// Step 4: Update sync state
const lastRecord = sourceRecords.records[sourceRecords.records.length - 1];
await ctx.step('update-sync-state', async () => {
await ctx.database.update({
database: 'sync-db',
event: 'update-sync-state',
where: { source: sourceSystem, target: targetSystem, entity: entityType },
data: {
lastSyncedAt: new Date().toISOString(),
lastSyncedId: lastRecord.id,
recordsSynced: ctx.getState('syncedRecords'),
conflictsFound: ctx.getState('conflicts')?.length || 0,
},
});
});
// Step 5: Report conflicts if any
const conflicts = ctx.getState('conflicts') || [];
if (conflicts.length > 0) {
await ctx.step(
'report-conflicts',
async () => {
await ctx.notification.email({
notification: 'system-alerts',
event: 'sync-conflicts',
recipients: [ctx.input.adminEmail],
template: {
sourceSystem,
targetSystem,
entityType,
conflictCount: conflicts.length,
conflicts,
},
});
},
null,
{ allow_fail: true }
);
}
return {
success: true,
synced: ctx.getState('syncedRecords'),
conflicts: conflicts.length,
hasMore: sourceRecords.hasMore,
};
},
});
ductape.features.define(Map.of(
"tag", "data-sync",
"name", "Data Synchronization",
options: Map.of(
"timeout", 600000,
"rollback_strategy", "to_checkpoint"
),
handler: async (ctx) => Map.of(
Map<String, Object> Map.of( sourceSystem, targetSystem, entityType, batchSize = 100 ) = ctx.input;
ctx.log.info('Starting data sync', Map.of( sourceSystem, targetSystem, entityType ));
// Step 1: Get last sync checkpoint
Map<String, Object> lastSync = ctx.step('get-last-sync', async () => Map.of(
Map<String, Object> result = ctx.database.query(Map.of(
"database", "sync-db",
"event", "get-sync-state",
params: Map.of( source: sourceSystem, target: targetSystem, entity: entityType )
));
return result[0] || Map.of( lastSyncedAt: null, lastSyncedId: null );
));
// Step 2: Fetch changed records from source
Map<String, Object> sourceRecords = ctx.step('fetch-source-records', async () => Map.of(
return ctx.api.run(Map.of(
app: sourceSystem,
"event", "get-records",
input: Map.of(
body: Map.of(
entityType,
modifiedAfter: lastSync.lastSyncedAt,
limit: batchSize
)
)
));
));
if (sourceRecords.records.length === 0) Map.of(
ctx.log.info('No records to sync');
return Map.of( "success", true, "synced", 0, "conflicts", 0 );
)
ctx.setState('totalRecords', sourceRecords.records.length);
ctx.setState('syncedRecords', 0);
ctx.setState('conflicts', []);
ctx.checkpoint('records-fetched', Map.of(
count: sourceRecords.records.length
));
// Step 3: Process each record
for (Map<String, Object> record of sourceRecords.records) Map.of(
ctx.step(`sync-record-$Map.of(record.id)`, async () => Map.of(
// Check for existing record in target
Map<String, Object> existing = ctx.api.run(Map.of(
app: targetSystem,
"event", "get-record",
input: Map.of( params: Map.of( id: record.id ) )
));
if (existing && existing.modifiedAt > record.modifiedAt) Map.of(
// Conflict: target is newer
Map<String, Object> conflicts = ctx.getState('conflicts') || [];
conflicts.push(Map.of(
recordId: record.id,
sourceModified: record.modifiedAt,
targetModified: existing.modifiedAt
));
ctx.setState('conflicts', conflicts);
ctx.log.warn('Sync conflict detected', Map.of( recordId: record.id ));
return Map.of( "skipped", true, "reason", "conflict" );
)
// Sync the record
if (existing) Map.of(
ctx.api.run(Map.of(
app: targetSystem,
"event", "update-record",
input: Map.of( body: record )
));
) else Map.of(
ctx.api.run(Map.of(
app: targetSystem,
"event", "create-record",
input: Map.of( body: record )
));
)
Map<String, Object> synced = (ctx.getState('syncedRecords') || 0) + 1;
ctx.setState('syncedRecords', synced);
return Map.of( "synced", true );
));
)
// Step 4: Update sync state
Map<String, Object> lastRecord = sourceRecords.records[sourceRecords.records.length - 1];
ctx.step('update-sync-state', async () => Map.of(
ctx.database.update(Map.of(
"database", "sync-db",
"event", "update-sync-state",
where: Map.of( source: sourceSystem, target: targetSystem, entity: entityType ),
data: Map.of(
lastSyncedAt: Instant.now().toISOString(),
lastSyncedId: lastRecord.id,
recordsSynced: ctx.getState('syncedRecords'),
conflictsFound: ctx.getState('conflicts')?.length || 0
)
));
));
// Step 5: Report conflicts if any
Map<String, Object> conflicts = ctx.getState('conflicts') || [];
if (conflicts.length > 0) Map.of(
ctx.step(
'report-conflicts',
async () => Map.of(
ctx.notification.email(Map.of(
"notification", "system-alerts",
"event", "sync-conflicts",
recipients: [ctx.input.adminEmail],
template: Map.of(
sourceSystem,
targetSystem,
entityType,
conflictCount: conflicts.length,
conflicts
)
));
),
null,
Map.of( "allow_fail", true )
);
)
return Map.of(
"success", true,
synced: ctx.getState('syncedRecords'),
conflicts: conflicts.length,
hasMore: sourceRecords.hasMore
);
)
));
client.features.define({
"tag": "data-sync",
"name": "Data Synchronization",
options: {
"timeout": 600000,
"rollback_strategy": "to_checkpoint",
},
handler: async (ctx) => {
const { sourceSystem, targetSystem, entityType, batchSize = 100 } = ctx.input;
ctx.log.info('Starting data sync', { sourceSystem, targetSystem, entityType });
// Step 1: Get last sync checkpoint
lastSync := ctx.step('get-last-sync', async () => {
result := ctx.database.query({
"database": "sync-db",
"event": "get-sync-state",
params: { source: sourceSystem, target: targetSystem, entity: entityType },
});
return result[0] || { lastSyncedAt: null, lastSyncedId: null };
});
// Step 2: Fetch changed records from source
sourceRecords := ctx.step('fetch-source-records', async () => {
return ctx.api.run({
app: sourceSystem,
"event": "get-records",
input: {
body: {
entityType,
modifiedAfter: lastSync.lastSyncedAt,
limit: batchSize,
},
},
});
});
if (sourceRecords.records.length === 0) {
ctx.log.info('No records to sync');
return { "success": true, "synced": 0, "conflicts": 0 };
}
ctx.setState('totalRecords', sourceRecords.records.length);
ctx.setState('syncedRecords', 0);
ctx.setState('conflicts', []);
ctx.checkpoint('records-fetched', {
count: sourceRecords.records.length,
});
// Step 3: Process each record
for (const record of sourceRecords.records) {
ctx.step(`sync-record-${record.id}`, async () => {
// Check for existing record in target
existing := ctx.api.run({
app: targetSystem,
"event": "get-record",
input: { params: { id: record.id } },
});
if (existing && existing.modifiedAt > record.modifiedAt) {
// Conflict: target is newer
conflicts := ctx.getState('conflicts') || [];
conflicts.push({
recordId: record.id,
sourceModified: record.modifiedAt,
targetModified: existing.modifiedAt,
});
ctx.setState('conflicts', conflicts);
ctx.log.warn('Sync conflict detected', { recordId: record.id });
return { "skipped": true, "reason": "conflict" };
}
// Sync the record
if (existing) {
ctx.api.run({
app: targetSystem,
"event": "update-record",
input: { body: record },
});
} else {
ctx.api.run({
app: targetSystem,
"event": "create-record",
input: { body: record },
});
}
synced := (ctx.getState('syncedRecords') || 0) + 1;
ctx.setState('syncedRecords', synced);
return { "synced": true };
});
}
// Step 4: Update sync state
lastRecord := sourceRecords.records[sourceRecords.records.length - 1];
ctx.step('update-sync-state', async () => {
ctx.database.update({
"database": "sync-db",
"event": "update-sync-state",
where: { source: sourceSystem, target: targetSystem, entity: entityType },
data: {
lastSyncedAt: new Date().toISOString(),
lastSyncedId: lastRecord.id,
recordsSynced: ctx.getState('syncedRecords'),
conflictsFound: ctx.getState('conflicts')?.length || 0,
},
});
});
// Step 5: Report conflicts if any
conflicts := ctx.getState('conflicts') || [];
if (conflicts.length > 0) {
ctx.step(
'report-conflicts',
async () => {
ctx.notification.email({
"notification": "system-alerts",
"event": "sync-conflicts",
recipients: [ctx.input.adminEmail],
template: {
sourceSystem,
targetSystem,
entityType,
conflictCount: conflicts.length,
conflicts,
},
});
},
null,
{ "allow_fail": true }
);
}
return {
"success": true,
synced: ctx.getState('syncedRecords'),
conflicts: conflicts.length,
hasMore: sourceRecords.hasMore,
};
},
});
await ductape.features.define({
["tag"] = "data-sync",
["name"] = "Data Synchronization",
options: {
["timeout"] = 600000,
["rollback_strategy"] = "to_checkpoint",
},
handler: async (ctx) => {
var { sourceSystem, targetSystem, entityType, batchSize = 100 } = ctx.input;
ctx.log.info('Starting data sync', { sourceSystem, targetSystem, entityType });
// Step 1: Get last sync checkpoint
var lastSync = await ctx.step('get-last-sync', async () => {
var result = await ctx.database.query({
["database"] = "sync-db",
["event"] = "get-sync-state",
params: { source: sourceSystem, target: targetSystem, entity: entityType },
});
return result[0] || { lastSyncedAt: null, lastSyncedId: null };
});
// Step 2: Fetch changed records from source
var sourceRecords = await ctx.step('fetch-source-records', async () => {
return ctx.api.run({
app: sourceSystem,
["event"] = "get-records",
input: {
body: {
entityType,
modifiedAfter: lastSync.lastSyncedAt,
limit: batchSize,
},
},
});
});
if (sourceRecords.records.length === 0) {
ctx.log.info('No records to sync');
return { ["success"] = true, ["synced"] = 0, ["conflicts"] = 0 };
}
ctx.setState('totalRecords', sourceRecords.records.length);
ctx.setState('syncedRecords', 0);
ctx.setState('conflicts', []);
await ctx.checkpoint('records-fetched', {
count: sourceRecords.records.length,
});
// Step 3: Process each record
for (var record of sourceRecords.records) {
await ctx.step(`sync-record-${record.id}`, async () => {
// Check for existing record in target
var existing = await ctx.api.run({
app: targetSystem,
["event"] = "get-record",
input: { params: { id: record.id } },
});
if (existing && existing.modifiedAt > record.modifiedAt) {
// Conflict: target is newer
var conflicts = ctx.getState('conflicts') || [];
conflicts.push({
recordId: record.id,
sourceModified: record.modifiedAt,
targetModified: existing.modifiedAt,
});
ctx.setState('conflicts', conflicts);
ctx.log.warn('Sync conflict detected', { recordId: record.id });
return { ["skipped"] = true, ["reason"] = "conflict" };
}
// Sync the record
if (existing) {
await ctx.api.run({
app: targetSystem,
["event"] = "update-record",
input: { body: record },
});
} else {
await ctx.api.run({
app: targetSystem,
["event"] = "create-record",
input: { body: record },
});
}
var synced = (ctx.getState('syncedRecords') || 0) + 1;
ctx.setState('syncedRecords', synced);
return { ["synced"] = true };
});
}
// Step 4: Update sync state
var lastRecord = sourceRecords.records[sourceRecords.records.length - 1];
await ctx.step('update-sync-state', async () => {
await ctx.database.update({
["database"] = "sync-db",
["event"] = "update-sync-state",
where: { source: sourceSystem, target: targetSystem, entity: entityType },
data: {
lastSyncedAt: DateTime.UtcNow.toISOString(),
lastSyncedId: lastRecord.id,
recordsSynced: ctx.getState('syncedRecords'),
conflictsFound: ctx.getState('conflicts')?.length || 0,
},
});
});
// Step 5: Report conflicts if any
var conflicts = ctx.getState('conflicts') || [];
if (conflicts.length > 0) {
await ctx.step(
'report-conflicts',
async () => {
await ctx.notification.email({
["notification"] = "system-alerts",
["event"] = "sync-conflicts",
recipients: [ctx.input.adminEmail],
template: {
sourceSystem,
targetSystem,
entityType,
conflictCount: conflicts.length,
conflicts,
},
});
},
null,
{ ["allow_fail"] = true }
);
}
return {
["success"] = true,
synced: ctx.getState('syncedRecords'),
conflicts: conflicts.length,
hasMore: sourceRecords.hasMore,
};
},
});
Tips for Building Features
1. Design for Failure
Always assume steps can fail. Define rollback handlers for critical operations:
- TypeScript
- Java
- Go
- .NET
await ctx.step('critical-op', doWork, undoWork, { critical: true });
ctx.step('critical-op', doWork, undoWork, Map.of( "critical", true ));
ctx.step('critical-op', doWork, undoWork, { "critical": true });
await ctx.step('critical-op', doWork, undoWork, { ["critical"] = true });
2. Use Checkpoints for Long Features
Break long features into phases:
- TypeScript
- Java
- Go
- .NET
await ctx.checkpoint('phase-1-complete');
ctx.checkpoint('phase-1-complete');
ctx.checkpoint('phase-1-complete');
await ctx.checkpoint('phase-1-complete');
3. Keep Steps Atomic
Each step should do one thing well:
- TypeScript
- Java
- Go
- .NET
// Good: Single responsibility
await ctx.step('charge-card', chargeCard);
await ctx.step('send-receipt', sendReceipt);
// Bad: Multiple responsibilities
await ctx.step('process-payment', async () => {
await chargeCard();
await sendReceipt();
await updateInventory();
});
// Good: Single responsibility
ctx.step('charge-card', chargeCard);
ctx.step('send-receipt', sendReceipt);
// Bad: Multiple responsibilities
ctx.step('process-payment', async () => Map.of(
chargeCard();
sendReceipt();
updateInventory();
));
// Good: Single responsibility
ctx.step('charge-card', chargeCard);
ctx.step('send-receipt', sendReceipt);
// Bad: Multiple responsibilities
ctx.step('process-payment', async () => {
chargeCard();
sendReceipt();
updateInventory();
});
// Good: Single responsibility
await ctx.step('charge-card', chargeCard);
await ctx.step('send-receipt', sendReceipt);
// Bad: Multiple responsibilities
await ctx.step('process-payment', async () => {
await chargeCard();
await sendReceipt();
await updateInventory();
});
4. Use allow_fail for Non-Critical Steps
- TypeScript
- Java
- Go
- .NET
await ctx.step('analytics', trackEvent, null, { allow_fail: true });
ctx.step('analytics', trackEvent, null, Map.of( "allow_fail", true ));
ctx.step('analytics', trackEvent, null, { "allow_fail": true });
await ctx.step('analytics', trackEvent, null, { ["allow_fail"] = true });
5. Implement Idempotency
Use idempotency keys for external operations:
- TypeScript
- Java
- Go
- .NET
await ctx.api.run({
app: 'stripe',
event: 'charge',
input: {
headers: { 'Idempotency-Key': `${ctx.feature_id}-payment` },
},
});
ctx.api.run(Map.of(
"app", "stripe",
"event", "charge",
input: Map.of(
headers: Map.of( 'Idempotency-Key': `$Map.of(ctx.feature_id)-payment` )
)
));
ctx.api.run({
"app": "stripe",
"event": "charge",
input: {
headers: { 'Idempotency-Key': `${ctx.feature_id}-payment` },
},
});
await ctx.api.run({
["app"] = "stripe",
["event"] = "charge",
input: {
headers: { 'Idempotency-Key': `${ctx.feature_id}-payment` },
},
});