germantax-e2e-mrg0smaa/workflow-wip.md

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

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

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

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

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

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

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)

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

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)

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.
})