340 lines
10 KiB
Markdown
340 lines
10 KiB
Markdown
# Workflow examples (WIP — blocked on pikku OSS bundler bug)
|
|
|
|
Three workflow patterns demonstrating graph DSL, sequential code-style with `workflow.sleep`, and complex (dynamic branching). All ready to drop back into the template once **pikkujs/pikku#553** lands (`pikku deploy plan` only emits bundle.js for ~half the units when workflows are present).
|
|
|
|
To resume:
|
|
1. Wait for #553 to be fixed in `@pikku/cli`.
|
|
2. Recreate the files below at the listed paths.
|
|
3. Sync template to Gitea, redeploy.
|
|
|
|
---
|
|
|
|
## Step functions
|
|
|
|
### `src/functions/enrich-card.function.ts`
|
|
|
|
```ts
|
|
import { z } from 'zod'
|
|
import { pikkuSessionlessFunc } from '#pikku'
|
|
|
|
export const EnrichCardInput = z.object({
|
|
title: z.string().min(1),
|
|
status: z.enum(['todo', 'doing', 'done']).default('todo'),
|
|
})
|
|
|
|
export const EnrichCardOutput = z.object({
|
|
title: z.string(),
|
|
status: z.string(),
|
|
priority: z.enum(['low', 'medium', 'high']),
|
|
labels: z.array(z.string()),
|
|
})
|
|
|
|
/**
|
|
* Heuristic enrichment — picks priority from title keywords and tags
|
|
* common categories. Stand-in for what would normally be an LLM call /
|
|
* lookup against project rules.
|
|
*/
|
|
export const enrichCard = pikkuSessionlessFunc({
|
|
input: EnrichCardInput,
|
|
output: EnrichCardOutput,
|
|
func: async ({ logger }, { title, status }) => {
|
|
const lower = title.toLowerCase()
|
|
let priority: 'low' | 'medium' | 'high' = 'medium'
|
|
if (/\b(urgent|asap|critical|p0|outage)\b/.test(lower)) priority = 'high'
|
|
else if (/\b(later|someday|maybe|nice to have)\b/.test(lower)) priority = 'low'
|
|
|
|
const labels: string[] = []
|
|
if (/\b(bug|fix|broken|error)\b/.test(lower)) labels.push('bug')
|
|
if (/\b(feat|feature|add)\b/.test(lower)) labels.push('feature')
|
|
if (/\b(doc|docs|readme)\b/.test(lower)) labels.push('docs')
|
|
|
|
logger.info(`enrichCard: '${title}' → priority=${priority} labels=[${labels.join(',')}]`)
|
|
return { title, status, priority, labels }
|
|
},
|
|
})
|
|
```
|
|
|
|
### `src/functions/notify-card-created.function.ts`
|
|
|
|
```ts
|
|
import { z } from 'zod'
|
|
import { pikkuSessionlessFunc } from '#pikku'
|
|
|
|
export const NotifyCardCreatedInput = z.object({
|
|
cardId: z.string(),
|
|
title: z.string(),
|
|
priority: z.enum(['low', 'medium', 'high']),
|
|
labels: z.array(z.string()),
|
|
})
|
|
|
|
/**
|
|
* Stand-in for Slack/email/webhook fan-out. High-priority cards would get
|
|
* an immediate ping, others batched. Logs only for now.
|
|
*/
|
|
export const notifyCardCreated = pikkuSessionlessFunc({
|
|
input: NotifyCardCreatedInput,
|
|
func: async ({ logger }, { cardId, title, priority, labels }) => {
|
|
const channel = priority === 'high' ? '#urgent' : '#kanban'
|
|
logger.info(
|
|
`notifyCardCreated: ${channel} card=${cardId} '${title}' priority=${priority} labels=[${labels.join(',')}]`,
|
|
)
|
|
},
|
|
})
|
|
```
|
|
|
|
### `src/functions/count-cards-by-status.function.ts`
|
|
|
|
```ts
|
|
import { z } from 'zod'
|
|
import { pikkuSessionlessFunc } from '#pikku'
|
|
|
|
export const CountCardsByStatusOutput = z.object({
|
|
todo: z.number(),
|
|
doing: z.number(),
|
|
done: z.number(),
|
|
})
|
|
|
|
export const countCardsByStatus = pikkuSessionlessFunc({
|
|
output: CountCardsByStatusOutput,
|
|
func: async ({ kysely }) => {
|
|
const rows = await kysely
|
|
.selectFrom('kanbanCard')
|
|
.select(({ fn }) => ['status', fn.countAll<number>().as('count')])
|
|
.groupBy('status')
|
|
.execute()
|
|
const out = { todo: 0, doing: 0, done: 0 } as Record<string, number>
|
|
for (const r of rows) out[r.status] = Number(r.count)
|
|
return out as { todo: number; doing: number; done: number }
|
|
},
|
|
})
|
|
```
|
|
|
|
### `src/functions/emit-digest.function.ts`
|
|
|
|
```ts
|
|
import { z } from 'zod'
|
|
import { pikkuSessionlessFunc } from '#pikku'
|
|
|
|
export const EmitDigestInput = z.object({
|
|
before: z.object({ todo: z.number(), doing: z.number(), done: z.number() }),
|
|
after: z.object({ todo: z.number(), doing: z.number(), done: z.number() }),
|
|
})
|
|
|
|
export const emitDigest = pikkuSessionlessFunc({
|
|
input: EmitDigestInput,
|
|
func: async ({ logger }, { before, after }) => {
|
|
const delta = {
|
|
todo: after.todo - before.todo,
|
|
doing: after.doing - before.doing,
|
|
done: after.done - before.done,
|
|
}
|
|
logger.info(
|
|
`emitDigest: before=${JSON.stringify(before)} after=${JSON.stringify(after)} delta=${JSON.stringify(delta)}`,
|
|
)
|
|
},
|
|
})
|
|
```
|
|
|
|
### `src/functions/get-card.function.ts`
|
|
|
|
```ts
|
|
import { z } from 'zod'
|
|
import { pikkuSessionlessFunc } from '#pikku'
|
|
import { NotFoundError } from '@pikku/core/errors'
|
|
|
|
export const GetCardInput = z.object({ cardId: z.string() })
|
|
|
|
export const GetCardOutput = z.object({
|
|
cardId: z.string(),
|
|
title: z.string(),
|
|
status: z.string(),
|
|
position: z.number(),
|
|
createdAt: z.string(),
|
|
})
|
|
|
|
export const getCard = pikkuSessionlessFunc({
|
|
input: GetCardInput,
|
|
output: GetCardOutput,
|
|
func: async ({ kysely }, { cardId }) => {
|
|
return await kysely
|
|
.selectFrom('kanbanCard')
|
|
.selectAll()
|
|
.where('cardId', '=', cardId)
|
|
.executeTakeFirstOrThrow(() => new NotFoundError(`Card ${cardId} not found`))
|
|
},
|
|
})
|
|
```
|
|
|
|
### `src/functions/move-card.function.ts`
|
|
|
|
```ts
|
|
import { z } from 'zod'
|
|
import { pikkuSessionlessFunc } from '#pikku'
|
|
import { NotFoundError } from '@pikku/core/errors'
|
|
|
|
export const MoveCardInput = z.object({
|
|
cardId: z.string(),
|
|
status: z.enum(['todo', 'doing', 'done', 'archived']),
|
|
})
|
|
|
|
export const MoveCardOutput = z.object({
|
|
cardId: z.string(),
|
|
fromStatus: z.string(),
|
|
toStatus: z.string(),
|
|
})
|
|
|
|
export const moveCard = pikkuSessionlessFunc({
|
|
input: MoveCardInput,
|
|
output: MoveCardOutput,
|
|
func: async ({ kysely, logger }, { cardId, status: toStatus }) => {
|
|
const current = await kysely
|
|
.selectFrom('kanbanCard')
|
|
.select('status')
|
|
.where('cardId', '=', cardId)
|
|
.executeTakeFirstOrThrow(() => new NotFoundError(`Card ${cardId} not found`))
|
|
|
|
await kysely
|
|
.updateTable('kanbanCard')
|
|
.set({ status: toStatus })
|
|
.where('cardId', '=', cardId)
|
|
.execute()
|
|
|
|
logger.info(`moveCard ${cardId}: ${current.status} → ${toStatus}`)
|
|
return { cardId, fromStatus: current.status, toStatus }
|
|
},
|
|
})
|
|
```
|
|
|
|
---
|
|
|
|
## Workflow wirings
|
|
|
|
### `src/wirings/card-onboarding.workflow.ts` — graph (DSL declarative)
|
|
|
|
```ts
|
|
import { pikkuWorkflowGraph } from '#pikku/workflow/pikku-workflow-types.gen.js'
|
|
|
|
/**
|
|
* GRAPH (DSL) workflow — card creation pipeline:
|
|
* enrich → createCard → notify
|
|
*
|
|
* Best fit for static DAGs where every step is known up-front and the
|
|
* runtime can introspect it (replay, fan-out UI, etc.).
|
|
*/
|
|
export const cardOnboardingWorkflow = pikkuWorkflowGraph({
|
|
description: 'Enrich a new card, persist it, then notify',
|
|
tags: ['kanban', 'onboarding'],
|
|
nodes: {
|
|
enrich: 'enrichCard',
|
|
create: 'createCard',
|
|
notify: 'notifyCardCreated',
|
|
},
|
|
config: {
|
|
enrich: {
|
|
next: 'create',
|
|
},
|
|
create: {
|
|
next: 'notify',
|
|
input: (ref) => ({
|
|
title: ref('enrich', 'title'),
|
|
status: ref('enrich', 'status'),
|
|
}),
|
|
},
|
|
notify: {
|
|
input: (ref) => ({
|
|
cardId: ref('create', 'cardId'),
|
|
title: ref('create', 'title'),
|
|
priority: ref('enrich', 'priority'),
|
|
labels: ref('enrich', 'labels'),
|
|
}),
|
|
},
|
|
},
|
|
})
|
|
```
|
|
|
|
### `src/wirings/board-digest.workflow.ts` — sequential code + `workflow.sleep`
|
|
|
|
```ts
|
|
import { pikkuWorkflowFunc } from '#pikku/workflow/pikku-workflow-types.gen.js'
|
|
|
|
/**
|
|
* DSL (sequential code) workflow — board snapshot before/after a 30s
|
|
* cool-off, then emit a digest with the delta.
|
|
*
|
|
* Best fit when steps include loops, conditionals, or `workflow.sleep`
|
|
* that don't map cleanly to a static graph. Each `workflow.do` step is
|
|
* still a real pikkuFunc — no inline business logic in the orchestrator.
|
|
*/
|
|
export const boardDigestWorkflow = pikkuWorkflowFunc<void, void>(
|
|
async (_services, _data, { workflow }) => {
|
|
const before = await workflow.do('snapshot before', 'countCardsByStatus', {})
|
|
|
|
// Pause to capture activity over a window. In real life this would be
|
|
// longer (e.g. 1h); 30s here keeps the example testable end-to-end.
|
|
await workflow.sleep('cool-off', 30_000)
|
|
|
|
const after = await workflow.do('snapshot after', 'countCardsByStatus', {})
|
|
|
|
await workflow.do('emit', 'emitDigest', { before, after })
|
|
},
|
|
)
|
|
```
|
|
|
|
### `src/wirings/card-auto-progress.workflow.ts` — complex (dynamic branching)
|
|
|
|
```ts
|
|
import { z } from 'zod'
|
|
import { pikkuWorkflowComplexFunc } from '#pikku/workflow/pikku-workflow-types.gen.js'
|
|
|
|
const Input = z.object({ cardId: z.string() })
|
|
|
|
/**
|
|
* COMPLEX workflow — promote a card automatically based on its current
|
|
* state and the title pattern. Branches dynamically inside the
|
|
* orchestrator: urgent + todo → fast-track to doing + ping; stale doing →
|
|
* nag; done → no-op. Each branch path can't be cleanly expressed as a
|
|
* static graph because the next-step set depends on data fetched at
|
|
* runtime.
|
|
*
|
|
* Note: prefer pikkuWorkflowGraph for everything that fits a static DAG.
|
|
* Use complex only when branching genuinely depends on inline data.
|
|
*/
|
|
export const cardAutoProgressWorkflow = pikkuWorkflowComplexFunc<
|
|
z.infer<typeof Input>,
|
|
void
|
|
>(async (_services, { cardId }, { workflow }) => {
|
|
const card = await workflow.do('fetch', 'getCard', { cardId })
|
|
|
|
const lower = card.title.toLowerCase()
|
|
const isUrgent = /\b(urgent|asap|critical|p0|outage)\b/.test(lower)
|
|
|
|
if (card.status === 'todo' && isUrgent) {
|
|
const moved = await workflow.do('promote', 'moveCard', { cardId, status: 'doing' })
|
|
await workflow.do('alert', 'notifyCardCreated', {
|
|
cardId,
|
|
title: card.title,
|
|
priority: 'high',
|
|
labels: ['auto-promoted'],
|
|
})
|
|
return
|
|
}
|
|
|
|
if (card.status === 'doing') {
|
|
// Cool-off — give it 60s in case it was just nudged. Then re-fetch
|
|
// and decide; if it moved during the wait, do nothing.
|
|
await workflow.sleep('cool-off', 60_000)
|
|
const fresh = await workflow.do('refetch', 'getCard', { cardId })
|
|
if (fresh.status === 'doing') {
|
|
await workflow.do('nag', 'notifyCardCreated', {
|
|
cardId,
|
|
title: card.title,
|
|
priority: 'medium',
|
|
labels: ['stale-in-progress'],
|
|
})
|
|
}
|
|
return
|
|
}
|
|
// todo (non-urgent) and done: no-op.
|
|
})
|
|
```
|