vynte-infra/plunk
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add better segmenting

Browse Source
This commit is contained in:
Dries Augustyns
2025-12-02 09:03:05 +01:00
parent b521a405bb
commit fe2e038037
23 changed files with 2285 additions and 855 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/controllers/Campaigns.ts
+8 -9
View File
@@ -8,7 +8,6 @@ import type {AuthResponse} from '../middleware/auth.js';
import {requireAuth} from '../middleware/auth.js';
import {CampaignService} from '../services/CampaignService.js';
import {DomainService} from '../services/DomainService.js';
import {type SegmentFilter} from '../services/SegmentService.js';
import {CatchAsync} from '../utils/asyncHandler.js';
@Controller('campaigns')
@@ -22,7 +21,7 @@ export class Campaigns {
@CatchAsync
private async create(req: Request, res: Response, next: NextFunction) {
const auth = res.locals.auth as AuthResponse;
const {name, description, subject, body, from, fromName, replyTo, audienceType, audienceFilter, segmentId} =
const {name, description, subject, body, from, fromName, replyTo, audienceType, audienceCondition, segmentId} =
CampaignSchemas.create.parse(req.body);
// Validate audience-specific fields
@@ -30,8 +29,8 @@ export class Campaigns {
throw new HttpException(400, 'Segment ID is required for SEGMENT audience type');
}
if (audienceType === CampaignAudienceType.FILTERED && !audienceFilter) {
throw new HttpException(400, 'Audience filter is required for FILTERED audience type');
if (audienceType === CampaignAudienceType.FILTERED && !audienceCondition) {
throw new HttpException(400, 'Audience condition is required for FILTERED audience type');
}
// Verify domain ownership and verification
@@ -46,7 +45,7 @@ export class Campaigns {
fromName,
replyTo,
audienceType,
audienceFilter: audienceFilter as SegmentFilter[] | undefined,
audienceCondition,
segmentId,
});
@@ -118,7 +117,7 @@ export class Campaigns {
private async update(req: Request, res: Response, next: NextFunction) {
const auth = res.locals.auth as AuthResponse;
const {id} = req.params;
const {name, description, subject, body, from, fromName, replyTo, audienceType, audienceFilter, segmentId} =
const {name, description, subject, body, from, fromName, replyTo, audienceType, audienceCondition, segmentId} =
req.body;
// Validate audience-specific fields if audienceType is being updated
@@ -126,8 +125,8 @@ export class Campaigns {
throw new HttpException(400, 'Segment ID is required for SEGMENT audience type');
}
if (audienceType === CampaignAudienceType.FILTERED && audienceFilter === undefined) {
throw new HttpException(400, 'Audience filter is required for FILTERED audience type');
if (audienceType === CampaignAudienceType.FILTERED && audienceCondition === undefined) {
throw new HttpException(400, 'Audience condition is required for FILTERED audience type');
}
// Verify domain ownership and verification if 'from' is being updated
@@ -144,7 +143,7 @@ export class Campaigns {
fromName,
replyTo,
audienceType,
audienceFilter,
audienceCondition,
segmentId,
});
apps/api/src/controllers/Contacts.ts
+4 -3
View File
@@ -47,6 +47,7 @@ export class Contacts {
/**
* GET /contacts/fields
* Get all available contact fields (both standard and custom fields from data JSON)
* Returns field names with inferred types (string, number, boolean, date)
*/
@Get('fields')
@Middleware([requireAuth])
@@ -55,11 +56,11 @@ export class Contacts {
const auth = res.locals.auth as AuthResponse;
try {
const fields = await ContactService.getAvailableFields(auth.projectId!);
const fieldsWithTypes = await ContactService.getAvailableFields(auth.projectId!);
return res.status(200).json({
fields,
count: fields.length,
fields: fieldsWithTypes,
count: fieldsWithTypes.length,
});
} catch (error) {
console.error('[CONTACTS] Failed to get available fields:', error);
apps/api/src/controllers/Segments.ts
+8 -8
View File
@@ -74,20 +74,20 @@ export class Segments {
@CatchAsync
public async create(req: Request, res: Response, next: NextFunction) {
const auth = res.locals.auth as AuthResponse;
const {name, description, filters, trackMembership} = req.body;
const {name, description, condition, trackMembership} = req.body;
if (!name) {
return res.status(400).json({error: 'Name is required'});
}
if (!filters || !Array.isArray(filters)) {
return res.status(400).json({error: 'Filters must be an array'});
if (!condition || typeof condition !== 'object') {
return res.status(400).json({error: 'Condition is required and must be an object'});
}
const segment = await SegmentService.create(auth.projectId!, {
name,
description,
filters,
condition,
trackMembership,
});
@@ -104,20 +104,20 @@ export class Segments {
public async update(req: Request, res: Response, next: NextFunction) {
const auth = res.locals.auth as AuthResponse;
const segmentId = req.params.id;
const {name, description, filters, trackMembership} = req.body;
const {name, description, condition, trackMembership} = req.body;
if (!segmentId) {
return res.status(400).json({error: 'Segment ID is required'});
}
if (filters !== undefined && !Array.isArray(filters)) {
return res.status(400).json({error: 'Filters must be an array'});
if (condition !== undefined && typeof condition !== 'object') {
return res.status(400).json({error: 'Condition must be an object'});
}
const segment = await SegmentService.update(auth.projectId!, segmentId, {
name,
description,
filters,
condition,
trackMembership,
});
apps/api/src/controllers/Workflows.ts
+20
View File
@@ -370,4 +370,24 @@ export class Workflows {
return res.status(200).json(execution);
}
/**
* POST /workflows/:id/executions/cancel-all
* Cancel all active executions for a workflow
*/
@Post(':id/executions/cancel-all')
@Middleware([requireAuth])
@CatchAsync
public async cancelAllExecutions(req: Request, res: Response, next: NextFunction) {
const auth = res.locals.auth as AuthResponse;
const workflowId = req.params.id;
if (!workflowId) {
return res.status(400).json({error: 'Workflow ID is required'});
}
const result = await WorkflowService.cancelAllExecutions(auth.projectId!, workflowId);
return res.status(200).json(result);
}
}
apps/api/src/services/CampaignService.ts
+25 -20
View File
@@ -1,5 +1,6 @@
import type {Campaign, Contact, Prisma} from '@plunk/db';
import {CampaignAudienceType, CampaignStatus} from '@plunk/db';
import type {FilterCondition} from '@plunk/types';
import {prisma} from '../database/prisma.js';
import {HttpException} from '../exceptions/index.js';
@@ -8,7 +9,7 @@ import {buildEmailFieldsUpdate} from '../utils/modelUpdate.js';
import {DomainService} from './DomainService.js';
import {EmailService} from './EmailService.js';
import {QueueService} from './QueueService.js';
import {type SegmentFilter, SegmentService} from './SegmentService.js';
import {SegmentService} from './SegmentService.js';
const BATCH_SIZE = 500; // Number of emails to process per batch (increased for better performance)
@@ -21,7 +22,7 @@ export interface CreateCampaignData {
fromName?: string;
replyTo?: string;
audienceType: CampaignAudienceType;
audienceFilter?: SegmentFilter[];
audienceCondition?: FilterCondition;
segmentId?: string;
}
@@ -34,7 +35,7 @@ export interface UpdateCampaignData {
fromName?: string;
replyTo?: string;
audienceType?: CampaignAudienceType;
audienceFilter?: SegmentFilter[];
audienceCondition?: FilterCondition;
segmentId?: string;
}
@@ -61,10 +62,10 @@ export class CampaignService {
}
}
// Validate filters if provided
if (data.audienceType === CampaignAudienceType.FILTERED && data.audienceFilter) {
// This will throw if filters are invalid
SegmentService.validateFilters(data.audienceFilter);
// Validate condition if provided
if (data.audienceType === CampaignAudienceType.FILTERED && data.audienceCondition) {
// This will throw if condition is invalid
SegmentService.validateCondition(data.audienceCondition);
}
// Create campaign
@@ -79,7 +80,7 @@ export class CampaignService {
fromName: data.fromName,
replyTo: data.replyTo,
audienceType: data.audienceType,
audienceFilter: (data.audienceFilter || null) as unknown as Prisma.InputJsonValue,
audienceCondition: (data.audienceCondition || null) as unknown as Prisma.InputJsonValue,
segmentId: data.segmentId,
status: CampaignStatus.DRAFT,
},
@@ -105,11 +106,11 @@ export class CampaignService {
updateData.audienceType = data.audienceType;
}
if (data.audienceFilter !== undefined) {
if (data.audienceFilter) {
SegmentService.validateFilters(data.audienceFilter);
if (data.audienceCondition !== undefined) {
if (data.audienceCondition) {
SegmentService.validateCondition(data.audienceCondition);
}
updateData.audienceFilter = (data.audienceFilter || null) as unknown as Prisma.InputJsonValue;
updateData.audienceCondition = (data.audienceCondition || null) as unknown as Prisma.InputJsonValue;
}
if (data.segmentId !== undefined) {
@@ -230,7 +231,7 @@ export class CampaignService {
fromName: campaign.fromName,
replyTo: campaign.replyTo,
audienceType: campaign.audienceType,
audienceFilter: campaign.audienceFilter as Prisma.InputJsonValue,
audienceCondition: campaign.audienceCondition as Prisma.InputJsonValue,
segmentId: campaign.segmentId,
status: CampaignStatus.DRAFT,
totalRecipients: 0,
@@ -587,14 +588,17 @@ export class CampaignService {
return this.buildSegmentWhereAsync(projectId, campaign.segmentId, baseWhere);
case CampaignAudienceType.FILTERED: {
const filters = campaign.audienceFilter as unknown as SegmentFilter[];
if (!filters || filters.length === 0) {
throw new HttpException(400, 'Audience filters are required for FILTERED audience type');
const condition = campaign.audienceCondition as unknown as FilterCondition;
if (!condition) {
throw new HttpException(400, 'Audience condition is required for FILTERED audience type');
}
// Use the SegmentService to build the where clause from the condition
const segmentWhere = SegmentService.buildConditionClause(condition);
return {
...baseWhere,
AND: filters.map(filter => SegmentService.buildFilterCondition(filter)),
...segmentWhere,
};
}
@@ -611,7 +615,7 @@ export class CampaignService {
segmentId: string,
baseWhere: Prisma.ContactWhereInput,
): Promise<Prisma.ContactWhereInput> {
// Fetch the segment to get its filters
// Fetch the segment to get its condition
const segment = await prisma.segment.findUnique({
where: {id: segmentId},
});
@@ -620,11 +624,12 @@ export class CampaignService {
throw new HttpException(404, 'Segment not found');
}
const filters = segment.filters as unknown as SegmentFilter[];
const condition = segment.condition as unknown as FilterCondition;
const segmentWhere = SegmentService.buildConditionClause(condition);
return {
...baseWhere,
AND: filters.map(filter => SegmentService.buildFilterCondition(filter)),
...segmentWhere,
};
}
apps/api/src/services/ContactService.ts
+67 -16
View File
@@ -330,29 +330,80 @@ export class ContactService {
/**
* Get all available contact fields for a project
* Returns both standard fields and custom fields from the data JSON column
* Now includes type information inferred from actual data
*
* @param projectId - The project ID to filter contacts
* @returns Array of field names (e.g., ["subscribed", "data.plan", "data.firstName"])
* @returns Array of field objects with name and type
*/
public static async getAvailableFields(projectId: string): Promise<string[]> {
// Standard fields
const standardFields = ['subscribed'];
public static async getAvailableFields(
projectId: string,
): Promise<Array<{field: string; type: 'string' | 'number' | 'boolean' | 'date'}>> {
// Standard fields with known types
const standardFields = [
{field: 'email', type: 'string' as const},
{field: 'subscribed', type: 'boolean' as const},
{field: 'createdAt', type: 'date' as const},
{field: 'updatedAt', type: 'date' as const},
];
// Get custom fields from the data JSON column
// Use raw SQL to extract all keys from the JSON data column
const result = await prisma.$queryRaw<Array<{key: string}>>`
SELECT DISTINCT jsonb_object_keys(data) as key
FROM contacts
WHERE
"projectId" = ${projectId}
AND data IS NOT NULL
AND jsonb_typeof(data) = 'object'
// Get custom fields from the data JSON column with type inference
// Use raw SQL to extract all keys and sample values from the JSON data column
const result = await prisma.$queryRaw<Array<{key: string; sample_value: string; json_type: string}>>`
WITH field_keys AS (
SELECT DISTINCT jsonb_object_keys(data) as key
FROM contacts
WHERE
"projectId" = ${projectId}
AND data IS NOT NULL
AND jsonb_typeof(data) = 'object'
),
field_samples AS (
SELECT
fk.key,
jsonb_typeof(c.data->fk.key) as json_type,
(c.data->>fk.key) as sample_value
FROM field_keys fk
CROSS JOIN LATERAL (
SELECT data
FROM contacts
WHERE
"projectId" = ${projectId}
AND data ? fk.key
AND data->fk.key IS NOT NULL
LIMIT 1
) c
)
SELECT
key,
sample_value,
json_type
FROM field_samples
`;
// Combine standard fields with custom fields (prefixed with "data.")
const customFields = result.map(row => `data.${row.key}`);
// Infer types from JSON types and sample values
const customFields = result.map(row => {
let type: 'string' | 'number' | 'boolean' | 'date' = 'string';
return [...standardFields, ...customFields].sort();
// PostgreSQL jsonb_typeof returns: "object", "array", "string", "number", "boolean", "null"
if (row.json_type === 'boolean') {
type = 'boolean';
} else if (row.json_type === 'number') {
type = 'number';
} else if (row.json_type === 'string' && row.sample_value) {
// Try to detect dates (ISO 8601 format)
const dateRegex = /^\d{4}-\d{2}-\d{2}(T\d{2}:\d{2}:\d{2}(\.\d{3})?Z?)?$/;
if (dateRegex.test(row.sample_value)) {
type = 'date';
}
}
return {
field: `data.${row.key}`,
type,
};
});
return [...standardFields, ...customFields].sort((a, b) => a.field.localeCompare(b.field));
}
/**
apps/api/src/services/SegmentService.ts
+317 -100
View File
@@ -1,27 +1,13 @@
import {type Contact, Prisma, type Segment} from '@plunk/db';
import type {FilterCondition, FilterGroup, SegmentFilter} from '@plunk/types';
import {prisma} from '../database/prisma.js';
import {HttpException} from '../exceptions/index.js';
import {EventService} from './EventService.js';
export interface SegmentFilter {
field: string; // e.g., "email", "data.plan", "subscribed"
operator:
| 'equals'
| 'notEquals'
| 'contains'
| 'notContains'
| 'greaterThan'
| 'lessThan'
| 'greaterThanOrEqual'
| 'lessThanOrEqual'
| 'exists'
| 'notExists'
| 'within'; // For date ranges
value?: unknown;
unit?: 'days' | 'hours' | 'minutes'; // For 'within' operator
}
// Re-export types for use in other services
export type {FilterCondition, FilterGroup, SegmentFilter} from '@plunk/types';
export interface PaginatedContacts {
contacts: Contact[];
@@ -77,7 +63,7 @@ export class SegmentService {
}
/**
* Get contacts that match a segment's filters
* Get contacts that match a segment's condition
*/
public static async getContacts(
projectId: string,
@@ -86,9 +72,9 @@ export class SegmentService {
pageSize = 20,
): Promise<PaginatedContacts> {
const segment = await this.get(projectId, segmentId);
const filters = segment.filters as unknown as SegmentFilter[];
const condition = segment.condition as unknown as FilterCondition;
const where = this.buildWhereClause(projectId, filters);
const where = this.buildWhereClause(projectId, condition);
const skip = (page - 1) * pageSize;
const [contacts, total] = await Promise.all([
@@ -118,15 +104,15 @@ export class SegmentService {
data: {
name: string;
description?: string;
filters: SegmentFilter[];
condition: FilterCondition;
trackMembership?: boolean;
},
): Promise<Segment> {
// Validate filters
this.validateFilters(data.filters);
// Validate condition
this.validateCondition(data.condition);
// Compute initial member count
const where = this.buildWhereClause(projectId, data.filters);
const where = this.buildWhereClause(projectId, data.condition);
const memberCount = await prisma.contact.count({where});
return prisma.segment.create({
@@ -134,7 +120,7 @@ export class SegmentService {
projectId,
name: data.name,
description: data.description,
filters: data.filters as unknown as Prisma.JsonArray,
condition: data.condition as unknown as Prisma.InputJsonValue,
trackMembership: data.trackMembership ?? false,
memberCount,
},
@@ -150,16 +136,16 @@ export class SegmentService {
data: {
name?: string;
description?: string;
filters?: SegmentFilter[];
condition?: FilterCondition;
trackMembership?: boolean;
},
): Promise<Segment> {
// First verify segment exists and belongs to project
await this.get(projectId, segmentId);
// Validate filters if provided
if (data.filters) {
this.validateFilters(data.filters);
// Validate condition if provided
if (data.condition) {
this.validateCondition(data.condition);
}
const updateData: Prisma.SegmentUpdateInput = {};
@@ -170,11 +156,11 @@ export class SegmentService {
if (data.description !== undefined) {
updateData.description = data.description;
}
if (data.filters !== undefined) {
updateData.filters = data.filters as unknown as Prisma.JsonArray;
if (data.condition !== undefined) {
updateData.condition = data.condition as unknown as Prisma.InputJsonValue;
// Recompute member count when filters change
const where = this.buildWhereClause(projectId, data.filters);
// Recompute member count when condition changes
const where = this.buildWhereClause(projectId, data.condition);
updateData.memberCount = await prisma.contact.count({where});
}
if (data.trackMembership !== undefined) {
@@ -222,8 +208,8 @@ export class SegmentService {
*/
public static async refreshMemberCount(projectId: string, segmentId: string): Promise<number> {
const segment = await this.get(projectId, segmentId);
const filters = segment.filters as unknown as SegmentFilter[];
const where = this.buildWhereClause(projectId, filters);
const condition = segment.condition as unknown as FilterCondition;
const where = this.buildWhereClause(projectId, condition);
const memberCount = await prisma.contact.count({where});
@@ -242,7 +228,7 @@ export class SegmentService {
public static async refreshAllMemberCounts(projectId: string): Promise<void> {
const segments = await prisma.segment.findMany({
where: {projectId},
select: {id: true, filters: true},
select: {id: true, condition: true},
});
// Process in batches to avoid overwhelming the database
@@ -253,8 +239,8 @@ export class SegmentService {
await Promise.all(
batch.map(async segment => {
try {
const filters = segment.filters as unknown as SegmentFilter[];
const where = this.buildWhereClause(projectId, filters);
const condition = segment.condition as unknown as FilterCondition;
const where = this.buildWhereClause(projectId, condition);
const memberCount = await prisma.contact.count({where});
await prisma.segment.update({
@@ -283,8 +269,8 @@ export class SegmentService {
throw new HttpException(400, 'Segment does not have membership tracking enabled');
}
const filters = segment.filters as unknown as SegmentFilter[];
const where = this.buildWhereClause(projectId, filters);
const condition = segment.condition as unknown as FilterCondition;
const where = this.buildWhereClause(projectId, condition);
// Get all matching contacts using cursor-based pagination to avoid memory issues
const BATCH_SIZE = 1000;
@@ -428,6 +414,18 @@ export class SegmentService {
public static buildFilterCondition(filter: SegmentFilter): Prisma.ContactWhereInput {
const {field, operator, value, unit} = filter;
// Handle event-based filters (e.g., "event.upgrade", "event.purchase")
if (field.startsWith('event.')) {
const eventName = field.substring(6); // Remove "event." prefix
return this.buildEventCondition(eventName, operator, value, unit);
}
// Handle email activity filters (e.g., "email.opened", "email.clicked")
if (field.startsWith('email.')) {
const activity = field.substring(6); // Remove "email." prefix
return this.buildEmailActivityCondition(activity, operator, value, unit);
}
// Handle JSON field paths (e.g., "data.plan")
if (field.startsWith('data.')) {
const jsonPath = field.substring(5); // Remove "data." prefix
@@ -449,78 +447,163 @@ export class SegmentService {
}
/**
* Validate segment filters
* Validate segment condition (recursive)
*/
public static validateFilters(filters: SegmentFilter[]): void {
if (!Array.isArray(filters)) {
throw new HttpException(400, 'Filters must be an array');
public static validateCondition(condition: FilterCondition): void {
if (!condition || typeof condition !== 'object') {
throw new HttpException(400, 'Condition must be an object');
}
if (filters.length === 0) {
throw new HttpException(400, 'At least one filter is required');
if (!condition.logic || !['AND', 'OR'].includes(condition.logic)) {
throw new HttpException(400, 'Condition logic must be either "AND" or "OR"');
}
for (const filter of filters) {
if (!filter.field) {
throw new HttpException(400, 'Filter field is required');
}
if (!Array.isArray(condition.groups) || condition.groups.length === 0) {
throw new HttpException(400, 'Condition must have at least one group');
}
if (!filter.operator) {
throw new HttpException(400, 'Filter operator is required');
}
const validOperators = [
'equals',
'notEquals',
'contains',
'notContains',
'greaterThan',
'lessThan',
'greaterThanOrEqual',
'lessThanOrEqual',
'exists',
'notExists',
'within',
];
if (!validOperators.includes(filter.operator)) {
throw new HttpException(400, `Invalid operator: ${filter.operator}`);
}
// Validate that operators that need a value have one
const operatorsNeedingValue = [
'equals',
'notEquals',
'contains',
'notContains',
'greaterThan',
'lessThan',
'greaterThanOrEqual',
'lessThanOrEqual',
'within',
];
if (operatorsNeedingValue.includes(filter.operator) && filter.value === undefined) {
throw new HttpException(400, `Operator "${filter.operator}" requires a value`);
}
// Validate unit for "within" operator
if (filter.operator === 'within' && !filter.unit) {
throw new HttpException(400, '"within" operator requires a unit (days, hours, or minutes)');
}
for (const group of condition.groups) {
this.validateGroup(group);
}
}
/**
* Build Prisma where clause from segment filters
* Validate filter group (recursive)
*/
private static buildWhereClause(projectId: string, filters: SegmentFilter[]): Prisma.ContactWhereInput {
const where: Prisma.ContactWhereInput = {
projectId,
AND: filters.map(filter => this.buildFilterCondition(filter)),
};
private static validateGroup(group: FilterGroup): void {
if (!group || typeof group !== 'object') {
throw new HttpException(400, 'Group must be an object');
}
return where;
if (!Array.isArray(group.filters)) {
throw new HttpException(400, 'Group filters must be an array');
}
// Groups can have filters, nested conditions, or both
const hasFilters = group.filters.length > 0;
const hasConditions = group.conditions !== undefined;
if (!hasFilters && !hasConditions) {
throw new HttpException(400, 'Group must have at least one filter or nested condition');
}
// Validate all filters in the group
for (const filter of group.filters) {
this.validateFilter(filter);
}
// Recursively validate nested conditions
if (group.conditions) {
this.validateCondition(group.conditions);
}
}
/**
* Validate individual filter
*/
private static validateFilter(filter: SegmentFilter): void {
if (!filter.field) {
throw new HttpException(400, 'Filter field is required');
}
if (!filter.operator) {
throw new HttpException(400, 'Filter operator is required');
}
const validOperators = [
'equals',
'notEquals',
'contains',
'notContains',
'greaterThan',
'lessThan',
'greaterThanOrEqual',
'lessThanOrEqual',
'exists',
'notExists',
'within',
'triggered',
'triggeredWithin',
'notTriggered',
];
if (!validOperators.includes(filter.operator)) {
throw new HttpException(400, `Invalid operator: ${filter.operator}`);
}
// Validate that operators that need a value have one
const operatorsNeedingValue = [
'equals',
'notEquals',
'contains',
'notContains',
'greaterThan',
'lessThan',
'greaterThanOrEqual',
'lessThanOrEqual',
'within',
'triggeredWithin',
];
if (operatorsNeedingValue.includes(filter.operator) && filter.value === undefined) {
throw new HttpException(400, `Operator "${filter.operator}" requires a value`);
}
// Validate unit for time-based operators
if (['within', 'triggeredWithin'].includes(filter.operator) && !filter.unit) {
throw new HttpException(400, `"${filter.operator}" operator requires a unit (days, hours, or minutes)`);
}
}
/**
* Build Prisma where clause from filter condition (entry point)
*/
private static buildWhereClause(projectId: string, condition: FilterCondition): Prisma.ContactWhereInput {
return {
projectId,
...this.buildConditionClause(condition),
};
}
/**
* Build Prisma clause from filter condition (recursive)
*/
public static buildConditionClause(condition: FilterCondition): Prisma.ContactWhereInput {
const groupClauses = condition.groups.map(group => this.buildGroupClause(group));
if (condition.logic === 'AND') {
return {AND: groupClauses};
} else {
return {OR: groupClauses};
}
}
/**
* Build Prisma clause from filter group (recursive)
*/
private static buildGroupClause(group: FilterGroup): Prisma.ContactWhereInput {
const clauses: Prisma.ContactWhereInput[] = [];
// Add filter conditions from this group
if (group.filters.length > 0) {
clauses.push(...group.filters.map(filter => this.buildFilterCondition(filter)));
}
// Add nested condition if present
if (group.conditions) {
clauses.push(this.buildConditionClause(group.conditions));
}
// All conditions within a group are combined with AND
if (clauses.length === 0) {
return {}; // Empty group returns empty where clause
}
if (clauses.length === 1) {
return clauses[0]!; // Safe to use non-null assertion since we checked length
}
return {AND: clauses};
}
/**
@@ -649,4 +732,138 @@ export class SegmentService {
throw new HttpException(400, `Unsupported time unit: ${unit}`);
}
}
/**
* Build condition for event-based filters
* Uses Prisma relations to efficiently query contacts who triggered specific events
*/
private static buildEventCondition(
eventName: string,
operator: string,
value: unknown,
unit?: 'days' | 'hours' | 'minutes',
): Prisma.ContactWhereInput {
switch (operator) {
case 'triggered':
// Contact has triggered this event at any time
return {
events: {
some: {
name: eventName,
},
},
};
case 'triggeredWithin': {
// Contact has triggered this event within the specified timeframe
if (!unit) {
throw new HttpException(400, 'Unit is required for "triggeredWithin" operator');
}
const now = new Date();
const milliseconds = this.getMilliseconds(value as number, unit);
const since = new Date(now.getTime() - milliseconds);
return {
events: {
some: {
name: eventName,
createdAt: {
gte: since,
},
},
},
};
}
case 'notTriggered':
// Contact has never triggered this event
return {
events: {
none: {
name: eventName,
},
},
};
default:
throw new HttpException(400, `Unsupported operator for event field: ${operator}`);
}
}
/**
* Build condition for email activity filters
* Uses Prisma relations to efficiently query contacts based on email engagement
*/
private static buildEmailActivityCondition(
activity: string,
operator: string,
value: unknown,
unit?: 'days' | 'hours' | 'minutes',
): Prisma.ContactWhereInput {
// Map activity names to Email model fields
const fieldMap: Record<string, string> = {
opened: 'openedAt',
clicked: 'clickedAt',
bounced: 'bouncedAt',
complained: 'complainedAt',
sent: 'sentAt',
delivered: 'deliveredAt',
};
const field = fieldMap[activity];
if (!field) {
throw new HttpException(400, `Unsupported email activity: ${activity}`);
}
switch (operator) {
case 'triggered':
// Contact has this email activity at any time
return {
emails: {
some: {
[field]: {
not: null,
},
},
},
};
case 'triggeredWithin': {
// Contact has this email activity within the specified timeframe
if (!unit) {
throw new HttpException(400, 'Unit is required for "triggeredWithin" operator');
}
const now = new Date();
const milliseconds = this.getMilliseconds(value as number, unit);
const since = new Date(now.getTime() - milliseconds);
return {
emails: {
some: {
[field]: {
gte: since,
},
},
},
};
}
case 'notTriggered':
// Contact has never had this email activity
return {
emails: {
none: {
[field]: {
not: null,
},
},
},
};
default:
throw new HttpException(400, `Unsupported operator for email activity field: ${operator}`);
}
}
}
apps/api/src/services/WorkflowExecutionService.ts
+22
View File
@@ -87,6 +87,17 @@ export class WorkflowExecutionService {
return;
}
// Check if workflow is disabled
// Note: We allow running executions to complete even if workflow is disabled
// This prevents disruption to contacts who are already in the workflow
// Only NEW executions are prevented when workflow is disabled (see startExecution in WorkflowService)
if (!execution.workflow.enabled) {
console.info(
`[WORKFLOW] Workflow ${execution.workflow.id} (${execution.workflow.name}) is disabled, but allowing execution ${executionId} to continue`,
);
// Allow execution to continue - no action needed
}
const step = execution.workflow.steps.find(s => s.id === stepId);
if (!step) {
throw new HttpException(404, 'Step not found in workflow');
@@ -135,6 +146,17 @@ export class WorkflowExecutionService {
return;
}
// Check if the workflow execution is now in WAITING state (DELAY steps do this)
// If so, the step has already been handled and queued - don't process next steps
const updatedExecution = await prisma.workflowExecution.findUnique({
where: {id: execution.id},
});
if (updatedExecution?.status === WorkflowExecutionStatus.WAITING) {
// Workflow is waiting (DELAY step has queued the next step) - don't process next steps now
return;
}
// Mark step as completed (for normal steps that complete immediately)
await prisma.workflowStepExecution.update({
where: {id: stepExecution.id},
apps/api/src/services/WorkflowService.ts
+264 -9
View File
@@ -31,6 +31,20 @@ export interface WorkflowExecutionWithDetails extends WorkflowExecution {
}
export class WorkflowService {
/**
* Check if a workflow has active executions
*/
private static async hasActiveExecutions(workflowId: string): Promise<number> {
return prisma.workflowExecution.count({
where: {
workflowId,
status: {
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
},
},
});
}
/**
* Get all workflows for a project with pagination
*/
@@ -175,7 +189,26 @@ export class WorkflowService {
},
): Promise<Workflow> {
// Verify workflow exists and belongs to project
await this.get(projectId, workflowId);
const workflow = await this.get(projectId, workflowId);
// Check if workflow is enabled and has active executions
if (workflow.enabled) {
const activeExecutions = await this.hasActiveExecutions(workflowId);
if (activeExecutions > 0) {
// Block changes to trigger configuration while executions are running
const hasCriticalChanges = data.triggerType !== undefined || data.triggerConfig !== undefined;
if (hasCriticalChanges) {
throw new HttpException(
409,
`Cannot modify workflow trigger while workflow has ${activeExecutions} active execution(s). ` +
'Please disable the workflow first or wait for executions to complete. ' +
'You can still update name, description, and re-entry settings.',
);
}
}
}
const updateData: Prisma.WorkflowUpdateInput = {};
@@ -299,7 +332,7 @@ export class WorkflowService {
},
): Promise<WorkflowStep> {
// First verify workflow belongs to project
await this.get(projectId, workflowId);
const workflow = await this.get(projectId, workflowId);
// Then verify step exists and belongs to workflow
const step = await prisma.workflowStep.findUnique({
@@ -310,6 +343,26 @@ export class WorkflowService {
throw new HttpException(404, 'Workflow step not found');
}
// Check if workflow is enabled and has active executions
if (workflow.enabled) {
const activeExecutions = await this.hasActiveExecutions(workflowId);
if (activeExecutions > 0) {
// Only allow safe changes: name and position updates
const hasCriticalChanges =
data.config !== undefined || data.templateId !== undefined;
if (hasCriticalChanges) {
throw new HttpException(
409,
`Cannot modify step configuration while workflow has ${activeExecutions} active execution(s). ` +
'Please disable the workflow first or wait for executions to complete. ' +
'You can still update the step name and position.',
);
}
}
}
const updateData: Prisma.WorkflowStepUpdateInput = {};
if (data.name !== undefined) updateData.name = data.name;
@@ -333,10 +386,13 @@ export class WorkflowService {
* Delete a workflow step
*/
public static async deleteStep(projectId: string, workflowId: string, stepId: string): Promise<void> {
await this.get(projectId, workflowId);
const workflow = await this.get(projectId, workflowId);
const step = await prisma.workflowStep.findUnique({
where: {id: stepId},
include: {
outgoingTransitions: true,
},
});
if (step?.workflowId !== workflowId) {
@@ -348,8 +404,117 @@ export class WorkflowService {
throw new HttpException(400, 'Cannot delete the trigger step. Every workflow must have a trigger.');
}
await prisma.workflowStep.delete({
where: {id: stepId},
// Check if workflow is enabled and has active executions on this step or downstream
if (workflow.enabled) {
// Check if any active executions are currently on this step
const executionsOnStep = await prisma.workflowExecution.count({
where: {
workflowId,
currentStepId: stepId,
status: {
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
},
},
});
if (executionsOnStep > 0) {
throw new HttpException(
409,
`Cannot delete step "${step.name}" while ${executionsOnStep} execution(s) are currently on this step. ` +
'Please disable the workflow first or wait for executions to complete.',
);
}
// Also check downstream steps for active executions
const allSteps = await prisma.workflowStep.findMany({
where: {workflowId},
include: {outgoingTransitions: true},
});
// Build adjacency map
const adjacencyMap = new Map<string, string[]>();
for (const s of allSteps) {
adjacencyMap.set(
s.id,
s.outgoingTransitions.map(t => t.toStepId),
);
}
// Find all downstream steps
const downstreamSteps = new Set<string>([stepId]);
const queue = [stepId];
while (queue.length > 0) {
const currentStepId = queue.shift()!;
const outgoingStepIds = adjacencyMap.get(currentStepId) || [];
for (const nextStepId of outgoingStepIds) {
if (!downstreamSteps.has(nextStepId)) {
downstreamSteps.add(nextStepId);
queue.push(nextStepId);
}
}
}
// Check if any active executions are on downstream steps
const executionsOnDownstream = await prisma.workflowExecution.count({
where: {
workflowId,
currentStepId: {in: Array.from(downstreamSteps)},
status: {
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
},
},
});
if (executionsOnDownstream > 0) {
throw new HttpException(
409,
`Cannot delete step "${step.name}" while ${executionsOnDownstream} execution(s) are on downstream steps. ` +
'Deleting this step would orphan those executions. ' +
'Please disable the workflow first or wait for executions to complete.',
);
}
}
// Find all downstream steps that need to be deleted (cascade)
// First, get all steps and transitions for this workflow to build a graph
const allSteps = await prisma.workflowStep.findMany({
where: {workflowId},
include: {outgoingTransitions: true},
});
// Build adjacency map for efficient traversal
const adjacencyMap = new Map<string, string[]>();
for (const s of allSteps) {
adjacencyMap.set(
s.id,
s.outgoingTransitions.map(t => t.toStepId),
);
}
// Use BFS to traverse the workflow graph and find all downstream steps
const stepsToDelete = new Set<string>([stepId]);
const queue = [stepId];
while (queue.length > 0) {
const currentStepId = queue.shift()!;
const outgoingStepIds = adjacencyMap.get(currentStepId) || [];
for (const nextStepId of outgoingStepIds) {
if (!stepsToDelete.has(nextStepId)) {
stepsToDelete.add(nextStepId);
queue.push(nextStepId);
}
}
}
// Delete all affected steps (Prisma will cascade delete the transitions)
await prisma.workflowStep.deleteMany({
where: {
id: {in: Array.from(stepsToDelete)},
workflowId, // Safety check to ensure we only delete steps from this workflow
},
});
}
@@ -424,6 +589,9 @@ export class WorkflowService {
* Delete a transition
*/
public static async deleteTransition(projectId: string, workflowId: string, transitionId: string): Promise<void> {
// Get workflow to check if it's enabled
const workflow = await this.get(projectId, workflowId);
// Verify transition exists and belongs to workflow
const transition = await prisma.workflowTransition.findFirst({
where: {
@@ -433,12 +601,70 @@ export class WorkflowService {
workflow: {projectId},
},
},
include: {
fromStep: true,
toStep: true,
},
});
if (!transition) {
throw new HttpException(404, 'Transition not found');
}
// Check if workflow is enabled and has active executions that could be affected
if (workflow.enabled) {
// Get all steps that would become orphaned by removing this transition
const allSteps = await prisma.workflowStep.findMany({
where: {workflowId},
include: {outgoingTransitions: true, incomingTransitions: true},
});
// Build adjacency map without this transition
const adjacencyMap = new Map<string, string[]>();
for (const s of allSteps) {
adjacencyMap.set(
s.id,
s.outgoingTransitions.filter(t => t.id !== transitionId).map(t => t.toStepId),
);
}
// Find all steps reachable from the toStep (downstream)
const downstreamSteps = new Set<string>([transition.toStepId]);
const queue = [transition.toStepId];
while (queue.length > 0) {
const currentStepId = queue.shift()!;
const outgoingStepIds = adjacencyMap.get(currentStepId) || [];
for (const nextStepId of outgoingStepIds) {
if (!downstreamSteps.has(nextStepId)) {
downstreamSteps.add(nextStepId);
queue.push(nextStepId);
}
}
}
// Check if any active executions are on the toStep or downstream steps
const executionsAffected = await prisma.workflowExecution.count({
where: {
workflowId,
currentStepId: {in: Array.from(downstreamSteps)},
status: {
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
},
},
});
if (executionsAffected > 0) {
throw new HttpException(
409,
`Cannot delete transition from "${transition.fromStep.name}" to "${transition.toStep.name}" ` +
`while ${executionsAffected} execution(s) are on affected steps. ` +
'Please disable the workflow first or wait for executions to complete.',
);
}
}
await prisma.workflowTransition.delete({
where: {id: transitionId},
});
@@ -646,26 +872,55 @@ export class WorkflowService {
data: {
status: WorkflowExecutionStatus.CANCELLED,
completedAt: new Date(),
exitReason: 'Cancelled by user',
},
});
}
/**
* Cancel all active executions for a workflow
*/
public static async cancelAllExecutions(
projectId: string,
workflowId: string,
): Promise<{cancelled: number}> {
// Verify workflow exists and belongs to project
await this.get(projectId, workflowId);
// Cancel all running and waiting executions
const result = await prisma.workflowExecution.updateMany({
where: {
workflowId,
status: {
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
},
},
data: {
status: WorkflowExecutionStatus.CANCELLED,
completedAt: new Date(),
exitReason: 'Cancelled by user (bulk cancel)',
},
});
return {cancelled: result.count};
}
/**
* Get all available fields for workflow conditions (contact fields + event fields)
*/
public static async getAvailableFields(projectId: string, eventName?: string) {
// Get contact fields (standard + custom data fields)
const contactFields = await ContactService.getAvailableFields(projectId);
const contactFieldsWithTypes = await ContactService.getAvailableFields(projectId);
// Add standard contact fields
const standardContactFields = ['contact.email', 'contact.subscribed'];
// Extract just the field names and prefix with 'contact.'
const contactFields = contactFieldsWithTypes.map(f => `contact.${f.field}`);
// Get event fields by analyzing actual event data
// This will only show fields that have been seen in actual events
const eventFields = await EventService.getAvailableEventFields(projectId, eventName);
// Combine all fields
const allFields = [...standardContactFields, ...contactFields, ...eventFields].sort();
const allFields = [...contactFields, ...eventFields].sort();
return {
fields: allFields,