Merge pull request #48 from parcelvoy/feat/journey-edit

WIP?: adds tags, journey step child, journey step save methods

db/migrations/20221210193723_add_tags.js

@@ -0,0 +1,44 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function(knex) {
+    return knex.schema
+        .createTable('tags', function(table) {
+            table.increments()
+            table.integer('project_id')
+                .unsigned()
+                .notNullable()
+                .references('id')
+                .inTable('projects')
+                .onDelete('CASCADE')
+            table.string('name', 255).defaultTo('')
+            table.timestamp('created_at').defaultTo(knex.fn.now())
+            table.timestamp('updated_at').defaultTo(knex.fn.now())
+        })
+        .createTable('entity_tags', function(table) {
+            table.increments()
+            table.integer('tag_id')
+                .unsigned()
+                .notNullable()
+                .references('id')
+                .inTable('tags')
+                .onDelete('CASCADE')
+            table.string('entity', 255)
+            table.integer('entity_id')
+                .unsigned()
+                .notNullable() // no fk constraint
+            table.timestamp('created_at').defaultTo(knex.fn.now())
+            table.timestamp('updated_at').defaultTo(knex.fn.now())
+        })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function(knex) {
+    return knex.schema
+        .dropTable('entity_tags')
+        .dropTable('tags')
+}

db/migrations/20230120144724_add_journey_step_uuid.js

@@ -0,0 +1,52 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = async function(knex) {
+    await knex.schema.alterTable('journey_steps', function(table) {
+        table.uuid('uuid').defaultTo()
+        table.integer('x')
+            .notNullable()
+            .defaultTo(0)
+            .alter()
+        table.integer('y')
+            .notNullable()
+            .defaultTo(0)
+            .alter()
+        table.unique(['journey_id', 'uuid'])
+    })
+    await knex.schema.createTable('journey_step_child', function(table) {
+        table.increments()
+        table.integer('step_id')
+            .unsigned()
+            .notNullable()
+            .references('id')
+            .inTable('journey_steps')
+            .onDelete('CASCADE')
+        table.integer('child_id')
+            .unsigned()
+            .notNullable()
+            .references('id')
+            .inTable('journey_steps')
+            .onDelete('CASCADE')
+        table.json('data')
+        table.timestamp('created_at').defaultTo(knex.fn.now())
+        table.timestamp('updated_at').defaultTo(knex.fn.now())
+        table.unique(['step_id', 'child_id'])
+    })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function(knex) {
+    return knex.schema
+        .alterTable('journey_steps', function(table) {
+            table.dropUnique(['journey_id', 'uuid'])
+            table.dropColumn('uuid')
+            table.integer('x').nullable().alter()
+            table.integer('y').nullable().alter()
+        })
+        .dropTable('journey_step_child')
+}

db/migrations/20230204200316_step_type_x_y_float.js

@@ -0,0 +1,23 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function(knex) {
+    return knex.schema
+        .alterTable('journey_steps', function(table) {
+            table.float('x').notNullable().defaultTo(0).alter()
+            table.float('y').notNullable().defaultTo(0).alter()
+        })
+}
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function(knex) {
+    return knex.schema
+        .alterTable('journey_steps', function(table) {
+            table.integer('x').notNullable().defaultTo(0).alter()
+            table.integer('y').notNullable().defaultTo(0).alter()
+        })
+}

src/campaigns/CampaignService.ts

@@ -17,12 +17,17 @@ import { getSubscription } from '../subscriptions/SubscriptionService'
 import { getProvider } from '../channels/ProviderRepository'
 import { isFuture } from 'date-fns'
 import { pick } from '../utilities'
+import { createTagSubquery } from '../tags/TagService'
 
 export const pagedCampaigns = async (params: SearchParams, projectId: number) => {
     return await Campaign.searchParams(
         params,
         ['name'],
-        b => b.where({ project_id: projectId }).whereNull('deleted_at'),
+        b => {
+            b.where({ project_id: projectId })
+            params?.tags?.length && b.whereIn('id', createTagSubquery(Campaign, projectId, params.tags))
+            return b
+        },
     )
 }
 

src/config/controllers.ts

@@ -12,6 +12,7 @@ import LinkController from '../render/LinkController'
 import TemplateController from '../render/TemplateController'
 import UserController from '../users/UserController'
 import ProfileController from '../profile/ProfileController'
+import TagController from '../tags/TagController'
 import { authMiddleware, scopeMiddleware } from '../auth/AuthMiddleware'
 import ProjectAdminController from '../projects/ProjectAdminController'
 
@@ -75,6 +76,7 @@ export const projectRouter = (prefix = '/projects/:project') => {
         ProviderController,
         ProjectAdminController,
         UserController,
+        TagController,
     )
 }
 

src/core/Model.ts

@@ -141,10 +141,10 @@ export default class Model {
         page = 0,
         itemsPerPage = 10,
         db: Database = App.main.db,
-    ): Promise<SearchResult<T>> {
+    ): Promise<SearchResult<InstanceType<T>>> {
         const total = await this.count(query, db)
         const start = page * itemsPerPage
-        const results = total > 0
+        const results: T[] = total > 0
             ? await query(this.table(db)).offset(start).limit(itemsPerPage)
             : []
         const end = Math.min(start + itemsPerPage, start + results.length)
@@ -232,7 +232,7 @@ export default class Model {
     ): Promise<InstanceType<T>> {
         const formattedData = this.formatJson(data)
         const id: number = await this.table(db).insert(formattedData)
-        return await this.find(id) as InstanceType<T>
+        return await this.find(id, b => b, db) as InstanceType<T>
     }
 
     static async update<T extends typeof Model>(
@@ -253,7 +253,7 @@ export default class Model {
     ): Promise<InstanceType<T>> {
         const formattedData = this.formatJson(data)
         await this.table(db).where('id', id).update(formattedData)
-        return await this.find(id) as InstanceType<T>
+        return await this.find(id, b => b, db) as InstanceType<T>
     }
 
     static async delete<T extends typeof Model>(

src/core/searchParams.ts

@@ -5,6 +5,7 @@ export interface SearchParams {
     itemsPerPage: number
     q?: string
     sort?: string
+    tags?: string[]
 }
 
 export const searchParamsSchema: JSONSchemaType<SearchParams> = {
@@ -30,5 +31,12 @@ export const searchParamsSchema: JSONSchemaType<SearchParams> = {
             type: 'string',
             nullable: true,
         },
+        tags: {
+            type: 'array',
+            items: {
+                type: 'string',
+            },
+            nullable: true,
+        },
     },
 }

src/journey/Journey.ts

@@ -1,5 +1,4 @@
 import Model, { ModelParams } from '../core/Model'
-import { JourneyStep } from './JourneyStep'
 
 export default class Journey extends Model {
     name!: string
@@ -7,8 +6,8 @@ export default class Journey extends Model {
     description?: string
     deleted_at?: Date
 
-    steps: JourneyStep[] = []
+    static virtualAttributes: string[] = ['steps']
 }
 
-export type JourneyParams = Omit<Journey, ModelParams | 'steps' | 'deleted_at'>
+export type JourneyParams = Omit<Journey, ModelParams | 'deleted_at'>
 export type UpdateJourneyParams = Omit<JourneyParams, 'project_id' | 'deleted_at'>

src/journey/JourneyController.ts

@@ -3,9 +3,9 @@ import { ProjectState } from '../auth/AuthMiddleware'
 import { searchParamsSchema } from '../core/searchParams'
 import { JSONSchemaType, validate } from '../core/validate'
 import { extractQueryParams } from '../utilities'
-import Journey, { JourneyParams, UpdateJourneyParams } from './Journey'
+import Journey, { JourneyParams } from './Journey'
-import { createJourney, createJourneyStep, deleteJourney, deleteJourneyStep, getJourney, pagedJourneys, updateJourney, updateJourneyStep } from './JourneyRepository'
+import { createJourney, deleteJourney, getJourneyStepMap, getJourney, pagedJourneys, setJourneyStepMap, updateJourney } from './JourneyRepository'
-import { JourneyStepParams } from './JourneyStep'
+import { JourneyStepMap, journeyStepTypes } from './JourneyStep'
 
 const router = new Router<
     ProjectState & { journey?: Journey }
@@ -41,7 +41,7 @@ router.post('/', async ctx => {
 
 router.param('journeyId', async (value, ctx, next) => {
     ctx.state.journey = await getJourney(parseInt(value), ctx.state.project.id)
-    if (!ctx.state.list) {
+    if (!ctx.state.journey) {
         ctx.throw(404)
         return
     }
@@ -52,25 +52,8 @@ router.get('/:journeyId', async ctx => {
     ctx.body = ctx.state.journey
 })
 
-const updateJourneyParams: JSONSchemaType<UpdateJourneyParams> = {
-    $id: 'updateJourneyParams',
-    type: 'object',
-    required: ['name'],
-    properties: {
-        name: {
-            type: 'string',
-        },
-        description: {
-            type: 'string',
-            nullable: true,
-        },
-    },
-    additionalProperties: false,
-}
-
 router.patch('/:journeyId', async ctx => {
-    const payload = validate(updateJourneyParams, ctx.request.body)
+    ctx.body = await updateJourney(ctx.state.journey!.id, validate(journeyParams, ctx.request.body))
-    ctx.body = await updateJourney(ctx.state.journey!.id, payload)
 })
 
 router.delete('/:journeyId', async ctx => {
@@ -78,18 +61,17 @@ router.delete('/:journeyId', async ctx => {
     ctx.body = true
 })
 
-const journeyStepParams: JSONSchemaType<JourneyStepParams> = {
+const journeyStepsParamsSchema: JSONSchemaType<JourneyStepMap> = {
-    $id: 'journeyStepParams',
+    $id: 'journeyStepsParams',
     type: 'object',
-    required: ['type'],
+    required: [],
+    additionalProperties: {
+        type: 'object',
+        required: ['type', 'x', 'y'],
         properties: {
             type: {
                 type: 'string',
-            enum: ['entrance', 'delay', 'action', 'gate', 'map'],
+                enum: Object.keys(journeyStepTypes),
-        },
-        child_id: {
-            type: 'integer',
-            nullable: true,
             },
             data: {
                 type: 'object', // TODO: Could validate further based on sub types
@@ -97,28 +79,40 @@ const journeyStepParams: JSONSchemaType<JourneyStepParams> = {
                 additionalProperties: true,
             },
             x: {
-            type: 'integer',
+                type: 'number',
             },
             y: {
-            type: 'integer',
+                type: 'number',
+            },
+            children: {
+                type: 'array',
+                nullable: true,
+                items: {
+                    type: 'object',
+                    required: ['uuid'],
+                    properties: {
+                        uuid: {
+                            type: 'string',
+                        },
+                        data: {
+                            type: 'object', // TODO: this is also specific to the parent node's type
+                            nullable: true,
+                            additionalProperties: true,
+                        },
+                    },
+                },
             },
         },
         additionalProperties: false,
+    },
 }
 
-router.post('/:journeyId/steps', async ctx => {
+router.get('/:journeyId/steps', async ctx => {
-    const payload = validate(journeyStepParams, ctx.request.body)
+    ctx.body = await getJourneyStepMap(ctx.state.journey!.id)
-    ctx.body = await createJourneyStep(ctx.state.journey!.id, payload)
 })
 
-router.patch('/:journeyId/steps/:stepId', async ctx => {
+router.put('/:journeyId/steps', async ctx => {
-    const payload = validate(journeyStepParams, ctx.request.body)
+    ctx.body = await setJourneyStepMap(ctx.state.journey!.id, validate(journeyStepsParamsSchema, ctx.request.body))
-    ctx.body = await updateJourneyStep(parseInt(ctx.params.id), payload)
-})
-
-router.delete('/:journeyId/steps/:stepId', async ctx => {
-    await deleteJourneyStep(parseInt(ctx.params.stepId))
-    ctx.body = true
 })
 
 export default router

src/journey/JourneyRepository.ts

@@ -1,7 +1,9 @@
+import App from '../app'
+import { Database } from 'config/database'
 import { RequestError } from '../core/errors'
 import { SearchParams } from '../core/searchParams'
 import Journey, { JourneyParams, UpdateJourneyParams } from './Journey'
-import { JourneyStep, JourneyEntrance, JourneyUserStep, JourneyStepParams } from './JourneyStep'
+import { JourneyStep, JourneyEntrance, JourneyUserStep, JourneyStepMap, toJourneyStepMap, JourneyStepChild } from './JourneyStep'
 
 export const pagedJourneys = async (params: SearchParams, projectId: number) => {
     return await Journey.searchParams(
@@ -16,20 +18,23 @@ export const allJourneys = async (projectId: number): Promise<Journey[]> => {
 }
 
 export const createJourney = async (projectId: number, params: JourneyParams): Promise<Journey> => {
-    return await Journey.insertAndFetch({
+    return App.main.db.transaction(async trx => {
+
+        const journey = await Journey.insertAndFetch({
             ...params,
             project_id: projectId,
-    })
+        }, trx)
 
-    // TODO: Should we create an entrance automatically here?
+        // auto-create entrance step
+        await JourneyEntrance.create(journey.id)
+
+        return journey
+    })
 }
 
 export const getJourney = async (id: number, projectId: number): Promise<Journey> => {
     const journey = await Journey.find(id, qb => qb.where('project_id', projectId))
     if (!journey) throw new RequestError('Journey not found', 404)
-
-    journey.steps = await allJourneySteps(journey.id)
-
     return journey
 }
 
@@ -41,8 +46,111 @@ export const deleteJourney = async (id: number): Promise<void> => {
     await Journey.updateAndFetch(id, { deleted_at: new Date() })
 }
 
-export const allJourneySteps = async (journeyId: number): Promise<JourneyStep[]> => {
+export const getJourneySteps = async (journeyId: number, db?: Database): Promise<JourneyStep[]> => {
-    return await JourneyStep.all(qb => qb.where('journey_id', journeyId))
+    return await JourneyStep.all(qb => qb.where('journey_id', journeyId), db)
+}
+
+export const getJourneyStepChildren = async (stepId: number) => {
+    return await JourneyStepChild.all(q => q.where('step_id', stepId))
+}
+
+const getAllJourneyStepChildren = async (journeyId: number, db?: Database): Promise<JourneyStepChild[]> => {
+    return await JourneyStepChild.all(
+        q => q.whereIn('step_id', JourneyStep.query(db).select('id').where('journey_id', journeyId)),
+        db,
+    )
+}
+
+export const getJourneyStepMap = async (journeyId: number) => {
+    const [steps, children] = await Promise.all([
+        getJourneySteps(journeyId),
+        getAllJourneyStepChildren(journeyId),
+    ])
+    return toJourneyStepMap(steps, children)
+}
+
+export const setJourneyStepMap = async (journeyId: number, stepMap: JourneyStepMap) => {
+    return await App.main.db.transaction(async trx => {
+
+        const [steps, children] = await Promise.all([
+            getJourneySteps(journeyId, trx),
+            getAllJourneyStepChildren(journeyId, trx),
+        ])
+
+        // Create or update steps
+        for (const [uuid, { type, x = 0, y = 0, data = {} }] of Object.entries(stepMap)) {
+            const idx = steps.findIndex(s => s.uuid === uuid)
+            if (idx === -1) {
+                steps.push(await JourneyStep.insertAndFetch({
+                    journey_id: journeyId,
+                    type,
+                    uuid,
+                    data,
+                    x,
+                    y,
+                }, trx))
+            } else {
+                const step = steps[idx]
+                steps[idx] = await JourneyStep.updateAndFetch(step.id, {
+                    type,
+                    uuid,
+                    data,
+                    x,
+                    y,
+                }, trx)
+            }
+        }
+
+        // Delete removed or unused steps
+        let i = 0
+        while (i < steps.length) {
+            const step = steps[i]
+            if (!stepMap[step.uuid]) {
+                await JourneyStep.delete(q => q.where('id', step.id), trx)
+                steps.splice(i, 1)
+                continue
+            }
+            i++
+        }
+
+        for (const step of steps) {
+            const list = stepMap[step.uuid]?.children ?? []
+            const childIds: number[] = []
+
+            for (const { uuid, data = {} } of list) {
+                const child = steps.find(s => s.uuid === uuid)
+                if (!child) continue
+                const idx = children.findIndex(sc => sc.step_id === step.id && sc.child_id === child.id)
+                let stepChild: JourneyStepChild
+                if (idx === -1) {
+                    children.push(stepChild = await JourneyStepChild.insertAndFetch({
+                        step_id: step.id,
+                        child_id: child.id,
+                        data,
+                    }, trx))
+                } else {
+                    stepChild = children[idx]
+                    children[idx] = await JourneyStepChild.updateAndFetch(stepChild.id, {
+                        data,
+                    }, trx)
+                }
+                childIds.push(stepChild.child_id)
+            }
+
+            i = 0
+            while (i < children.length) {
+                const stepChild = children[i]
+                if (stepChild.step_id === step.id && !childIds.includes(stepChild.child_id)) {
+                    await JourneyStepChild.delete(q => q.where('id', stepChild.id), trx)
+                    children.splice(i, 1)
+                    continue
+                }
+                i++
+            }
+        }
+
+        return toJourneyStepMap(steps, children)
+    })
 }
 
 export const getJourneyStep = async (id?: number): Promise<JourneyStep | undefined> => {
@@ -50,18 +158,6 @@ export const getJourneyStep = async (id?: number): Promise<JourneyStep | undefin
     return await JourneyStep.first(db => db.where('id', id))
 }
 
-export const createJourneyStep = async (journeyId: number, params: JourneyStepParams): Promise<JourneyStep> => {
-    return await JourneyStep.insertAndFetch({ ...params, journey_id: journeyId })
-}
-
-export const updateJourneyStep = async (id: number, params: JourneyStepParams): Promise<JourneyStep> => {
-    return await JourneyStep.updateAndFetch(id, { ...params })
-}
-
-export const deleteJourneyStep = async (id: number): Promise<number> => {
-    return await JourneyStep.delete(qb => qb.where('id', id))
-}
-
 export const getUserJourneyIds = async (userId: number): Promise<number[]> => {
     return await JourneyUserStep.all(
         db => db.where('user_id', userId)

src/journey/JourneyService.ts

@@ -1,6 +1,6 @@
 import { User } from '../users/User'
 import { getJourneyEntrance, getJourneyStep, getUserJourneyIds, lastJourneyStep } from './JourneyRepository'
-import { JourneyEntrance, JourneyDelay, JourneyGate, JourneyStep, JourneyMap, JourneyAction } from './JourneyStep'
+import { JourneyEntrance, JourneyStep, journeyStepTypes } from './JourneyStep'
 import { UserEvent } from '../users/UserEvent'
 import List from '../lists/List'
 
@@ -42,19 +42,26 @@ export default class JourneyService {
 
     async run(user: User, event?: UserEvent): Promise<void> {
 
+        const processed: number[] = []
+
         // Loop through all possible next steps until we get an empty next
         // which signifies that the journey is in a pending state
 
         let nextStep: JourneyStep | undefined | null = await this.nextStep(user)
         while (nextStep) {
-            const parsedStep = this.parse(nextStep)
+            if (processed.includes(nextStep.id)) {
+                // Avoid infinite loop in single run
+                break
+            }
+            processed.push(nextStep.id)
+            nextStep = this.parse(nextStep)
 
             // If completed, jump to next otherwise validate condition
-            if (await parsedStep.hasCompleted(user)) {
+            if (await nextStep.hasCompleted(user)) {
-                nextStep = await parsedStep.next(user)
+                nextStep = await nextStep.next(user)
-            } else if (await parsedStep.condition(user, event)) {
+            } else if (await nextStep.condition(user, event)) {
-                await parsedStep.complete(user)
+                await nextStep.complete(user)
-                nextStep = await parsedStep.next(user)
+                nextStep = await nextStep.next(user)
             } else {
                 nextStep = null
             }
@@ -62,14 +69,7 @@ export default class JourneyService {
     }
 
     parse(step: JourneyStep): JourneyStep {
-        const options = {
+        return journeyStepTypes[step.type]?.fromJson(step)
-            [JourneyEntrance.type]: JourneyEntrance,
-            [JourneyDelay.type]: JourneyDelay,
-            [JourneyGate.type]: JourneyGate,
-            [JourneyMap.type]: JourneyMap,
-            [JourneyAction.type]: JourneyAction,
-        }
-        return options[step.type]?.fromJson(step)
     }
 
     async nextStep(user: User): Promise<JourneyStep | undefined> {

src/journey/JourneyStep.ts

@@ -3,10 +3,12 @@ import Model from '../core/Model'
 import { User } from '../users/User'
 import Rule from '../rules/Rule'
 import { check } from '../rules/RuleEngine'
-import { getJourneyStep, getUserJourneyStep } from './JourneyRepository'
+import { getJourneyStep, getJourneyStepChildren, getUserJourneyStep } from './JourneyRepository'
 import { UserEvent } from '../users/UserEvent'
 import { getCampaign, sendCampaign } from '../campaigns/CampaignService'
-import { snakeCase } from '../utilities'
+import { random, snakeCase, uuid } from '../utilities'
+import App from '../app'
+import JourneyProcessJob from './JourneyProcessJob'
 
 export class JourneyUserStep extends Model {
     user_id!: number
@@ -17,11 +19,22 @@ export class JourneyUserStep extends Model {
     static tableName = 'journey_user_step'
 }
 
+export class JourneyStepChild extends Model {
+
+    step_id!: number
+    child_id!: number
+    data?: Record<string, unknown>
+
+    static tableName = 'journey_step_child'
+    static jsonAttributes: string[] = ['data']
+
+}
+
 export class JourneyStep extends Model {
     type!: string
     journey_id!: number
-    child_id?: number // the step that comes after
     data?: Record<string, unknown>
+    uuid!: string
 
     // UI variables
     x = 0
@@ -63,12 +76,12 @@ export class JourneyStep extends Model {
      */
     // eslint-disable-next-line @typescript-eslint/no-unused-vars
     async next(user: User, event?: UserEvent): Promise<JourneyStep | undefined> {
-        return await getJourneyStep(this.child_id)
+        const child = await JourneyStepChild.first(q => q.where('step_id', this.id))
+        if (!child) return undefined
+        return await getJourneyStep(child.child_id)
     }
 }
 
-export type JourneyStepParams = Pick<JourneyStep, 'type' | 'child_id' | 'data' | 'x' | 'y'>
-
 export class JourneyEntrance extends JourneyStep {
     static type = 'entrance'
 
@@ -79,14 +92,16 @@ export class JourneyEntrance extends JourneyStep {
         this.list_id = json?.data.list_id
     }
 
-    static async create(listId: number, childId?: number, journeyId?: number): Promise<JourneyEntrance> {
+    static async create(journeyId: number, listId?: number): Promise<JourneyEntrance> {
         return await JourneyEntrance.insertAndFetch({
             type: this.type,
-            child_id: childId,
+            uuid: uuid(),
             journey_id: journeyId,
             data: {
                 list_id: listId,
             },
+            x: 0,
+            y: 0,
         })
     }
 }
@@ -155,23 +170,19 @@ export class JourneyAction extends JourneyStep {
 export class JourneyGate extends JourneyStep {
     static type = 'gate'
 
-    entrance_type!: 'user' | 'event'
     rule!: Rule
 
     parseJson(json: any) {
         super.parseJson(json)
 
-        this.entrance_type = json?.data?.entrance_type
         this.rule = json?.data.rule
     }
 
-    static async create(entranceType: string, rule: Rule, childId?: number, journeyId?: number): Promise<JourneyGate> {
+    static async create(rule: Rule, journeyId?: number): Promise<JourneyGate> {
         return await JourneyGate.insertAndFetch({
             type: this.type,
-            child_id: childId,
             journey_id: journeyId,
             data: {
-                entrance_type: entranceType,
                 rule,
             },
         })
@@ -191,13 +202,11 @@ export class JourneyMap extends JourneyStep {
     static type = 'map'
 
     attribute!: string
-    options!: MapOptions
 
     parseJson(json: any) {
         super.parseJson(json)
 
         this.attribute = json?.data?.attribute
-        this.options = json?.data.options
     }
 
     static async create(attribute: string, options: MapOptions, journeyId?: number): Promise<JourneyMap> {
@@ -217,13 +226,138 @@ export class JourneyMap extends JourneyStep {
 
     async next(user: User) {
 
+        const children = await getJourneyStepChildren(this.id)
+
         // When comparing the user expects comparison
         // to be between the UI model user vs core user
         const templateUser = user.flatten()
 
         // Based on an attribute match, pick a child step
         const value = templateUser[this.attribute]
-        const childId = this.options[value]
+        for (const { child_id, data = {} } of children) {
-        return await getJourneyStep(childId)
+            if (data.value === value) {
+                return await getJourneyStep(child_id)
+            }
+        }
+
+        // Fallback
+        for (const { child_id, data } of children) {
+            if (data?.fallback) {
+                return await getJourneyStep(child_id)
+            }
+        }
+
+        return undefined
     }
 }
+
+/**
+ * randomly distribute users to different branches
+ */
+export class JourneyExperiment extends JourneyStep {
+    static type = 'experiment'
+
+    async next() {
+
+        let children = await getJourneyStepChildren(this.id)
+
+        if (!children.length) return undefined
+
+        children = children.reduce<JourneyStepChild[]>((a, c) => {
+            const proportion = Number(c.data?.value)
+            if (!isNaN(proportion) && proportion > 0) {
+                for (let i = 0; i < proportion; i++) {
+                    a.push(c)
+                }
+            }
+            return a
+        }, [])
+
+        if (!children) return undefined
+
+        return await getJourneyStep(random(children).child_id)
+    }
+
+}
+
+/**
+ * add user to another journey
+ */
+export class JourneyLink extends JourneyStep {
+    static type = 'link'
+
+    target_id!: number
+
+    parseJson(json: any) {
+        super.parseJson(json)
+        this.target_id = json.data?.journey_id
+    }
+
+    async complete(user: User, event?: UserEvent | undefined): Promise<void> {
+
+        if (!isNaN(this.journey_id)) {
+            await App.main.queue.enqueue(JourneyProcessJob.from({
+                journey_id: this.target_id,
+                user_id: user.id,
+                event_id: event?.id,
+            }))
+        }
+
+        return super.complete(user, event)
+    }
+}
+
+export const journeyStepTypes = [
+    JourneyEntrance,
+    JourneyDelay,
+    JourneyAction,
+    JourneyGate,
+    JourneyMap,
+    JourneyExperiment,
+    JourneyLink,
+].reduce<Record<string, typeof JourneyStep>>((a, c) => {
+    a[c.type] = c
+    return a
+}, {})
+
+export type JourneyStepMap = Record<
+    string,
+    {
+        type: string
+        data?: Record<string, unknown>
+        x: number
+        y: number
+        children?: Array<{
+            uuid: string
+            data?: Record<string, unknown>
+        }>
+    }
+>
+
+// This is async in case we ever want to fetch stats here
+export async function toJourneyStepMap(steps: JourneyStep[], children: JourneyStepChild[]) {
+    const editData: JourneyStepMap = {}
+
+    for (const step of steps) {
+        editData[step.uuid] = {
+            type: step.type,
+            data: step.data ?? {},
+            x: step.x ?? 0,
+            y: step.y ?? 0,
+            children: children.reduce<JourneyStepMap[string]['children']>((a, { step_id, child_id, data }) => {
+                if (step_id === step.id) {
+                    const child = steps.find(s => s.id === child_id)
+                    if (child) {
+                        a!.push({
+                            uuid: child.uuid,
+                            data,
+                        })
+                    }
+                }
+                return a
+            }, []),
+        }
+    }
+
+    return editData
+}

src/journey/__test__/JourneyService.spec.ts

@@ -4,7 +4,7 @@ import { User } from '../../users/User'
 import Journey from '../Journey'
 import { lastJourneyStep } from '../JourneyRepository'
 import JourneyService from '../JourneyService'
-import { JourneyEntrance, JourneyMap } from '../JourneyStep'
+import { JourneyEntrance, JourneyMap, JourneyStepChild } from '../JourneyStep'
 
 describe('Run', () => {
     describe('step progression', () => {
@@ -26,7 +26,12 @@ describe('Run', () => {
                 US: 43,
                 ES: 34,
             }, journey.id)
-            await JourneyEntrance.create(list.id, step2.id, journey.id)
+            const entrance = await JourneyEntrance.create(journey.id, list.id)
+            await JourneyStepChild.insert({
+                step_id: entrance.id,
+                child_id: step2.id,
+                data: {},
+            })
 
             const service = new JourneyService(journey.id)
 

src/journey/__test__/JourneyStep.spec.ts

@@ -1,7 +1,7 @@
 import { createProject } from '../../projects/ProjectService'
 import Rule from '../../rules/Rule'
 import { User } from '../../users/User'
-import { JourneyStep, JourneyMap, JourneyGate } from '../JourneyStep'
+import { JourneyStep, JourneyMap, JourneyGate, JourneyStepChild } from '../JourneyStep'
 
 describe('JourneyGate', () => {
 
@@ -13,7 +13,7 @@ describe('JourneyGate', () => {
 
     describe('user gate', () => {
         const createGate = async (rule: Rule) => {
-            return await JourneyGate.create('user', rule)
+            return await JourneyGate.create(rule)
         }
 
         test('condition passes with valid equals rule', async () => {
@@ -80,7 +80,7 @@ describe('JourneyGate', () => {
 describe('Journey Map', () => {
     test('different options pick different paths', async () => {
 
-        const user = User.fromJson({ data: { progress: '20' } })
+        const user = User.fromJson({ data: { progress: 20 } })
         const step1 = await JourneyStep.insertAndFetch()
         const step2 = await JourneyStep.insertAndFetch()
         const step3 = await JourneyStep.insertAndFetch()
@@ -90,10 +90,21 @@ describe('Journey Map', () => {
             20: step2.id,
             30: step3.id,
         })
+
+        await Promise.all(([
+            [10, step1],
+            [20, step2],
+            [30, step3],
+        ] as const).map(([value, { id: child_id }]) => JourneyStepChild.insert({
+            step_id: map.id,
+            child_id,
+            data: { value },
+        })))
+
         const value1 = await map.next(user)
         expect(value1?.id).toEqual(step2.id)
 
-        user.data.progress = '30'
+        user.data.progress = 30
         const value2 = await map.next(user)
         expect(value2?.id).toEqual(step3.id)
     })

src/tags/Tag.ts

@@ -0,0 +1,14 @@
+import Model, { ModelParams } from '../core/Model'
+
+export class Tag extends Model {
+    project_id!: number
+    name!: string
+}
+
+export class EntityTag extends Model {
+    entity!: string // table name
+    entity_id!: number
+    tag_id!: number
+}
+
+export type TagParams = Omit<Tag, ModelParams | 'project_id'>

src/tags/TagController.ts

@@ -0,0 +1,62 @@
+import Router from '@koa/router'
+import { JSONSchemaType } from 'ajv'
+import { searchParamsSchema } from '../core/searchParams'
+import { validate } from '../core/validate'
+import { extractQueryParams } from '../utilities'
+import { ProjectState } from '../auth/AuthMiddleware'
+import { Tag, TagParams } from './Tag'
+
+const router = new Router<
+    ProjectState & {
+        tag?: Tag
+    }
+>({
+    prefix: '/tags',
+})
+
+router.get('/', async ctx => {
+    ctx.body = await Tag.searchParams(
+        extractQueryParams(ctx.request.query, searchParamsSchema),
+        ['name'],
+    )
+})
+
+const tagParams: JSONSchemaType<TagParams> = {
+    type: 'object',
+    required: ['name'],
+    properties: {
+        name: {
+            type: 'string',
+        },
+    },
+}
+
+router.post('/', async ctx => {
+    ctx.body = await Tag.insertAndFetch(validate(tagParams, ctx.request.body))
+})
+
+router.param('tagId', async (value, ctx, next) => {
+    ctx.state.tag = await Tag.first(b => b.where({
+        project_id: ctx.state.project.id,
+        id: value,
+    }))
+    if (!ctx.state.tag) {
+        return ctx.throw(404)
+    }
+    return await next()
+})
+
+router.get('/:tagId', async ctx => {
+    ctx.body = ctx.state.tag!
+})
+
+router.patch('/:tagId', async ctx => {
+    ctx.body = await Tag.updateAndFetch(ctx.state.tag!.id, validate(tagParams, ctx.request.body))
+})
+
+router.delete('/:tagId', async ctx => {
+    await Tag.delete(b => b.where('id', ctx.state.tag!.id))
+    ctx.body = true
+})
+
+export default router

src/tags/TagService.ts

@@ -0,0 +1,67 @@
+import Model from 'core/Model'
+import { EntityTag, Tag } from './Tag'
+
+// use transaction?
+export async function setTags<T extends Model & { project_id: number }>(target: T, names: string[]) {
+
+    // is there a better way to do this?
+    const tableName = (Object.getPrototypeOf(target) as typeof Model).tableName
+    const { project_id } = target
+
+    // if empty value passed, remove all tag relations
+    if (!names?.length) {
+        return await EntityTag.delete(b => b.where({
+            entity: tableName,
+            entity_id: target.id,
+        }))
+    }
+
+    // find/create tags in this project by name
+    const tags = await Tag.all(b => b.where({
+        project_id,
+        names,
+    }))
+
+    for (const name of names) {
+        if (!tags.find(t => t.name === name)) {
+            tags.push(await Tag.insertAndFetch({
+                project_id,
+                name,
+            }))
+        }
+    }
+
+    const relations = await EntityTag.all(b => b.where({
+        entity: tableName,
+        entity_id: target.id,
+    }))
+
+    for (const tag of tags) {
+        if (!relations.find(r => r.tag_id === tag.id)) {
+            await EntityTag.insert({
+                entity: tableName,
+                entity_id: target.id,
+                tag_id: tag.id,
+            })
+        }
+    }
+
+    const remove = relations.filter(r => !tags.find(t => t.id === r.tag_id)).map(r => r.id)
+    if (remove.length) {
+        await EntityTag.delete(b => b.whereIn('id', remove))
+    }
+
+    return names
+}
+
+// use with knex:  myQuery.whereIn('id', createTagSubquery(MyEntity, 1, ['tag 1', 'tag 2']))
+export function createTagSubquery<T extends typeof Model>(model: T, project_id: number, names: string[]) {
+    return EntityTag.query()
+        .select('entity_id')
+        .join('tags', 'tag_id', 'tag.id')
+        .whereIn('tag.name', names)
+        .andWhere('tag.project_id', project_id)
+        .andWhere('entity', model.tableName)
+        .groupBy('tag.id')
+        .having(model.raw(`count(*) > ${names.length}`))
+}