<html lang="en">
<head></head>
<body>

<form id="mainForm" method="post" action="https://stackblitz.com/run" target="_self">
<input type="hidden" name="project[files][README.md]" value="# Cloudflare D1 Workflow POC

This demo validates TanStack Workflow with:

- Cloudflare Workers HTTP routes
- Cloudflare Cron Triggers
- Cloudflare D1 as the durable `WorkflowExecutionStore`
- package-owned Workflow store migrations

It uses the runtime/store path:

```ts
createCloudflareD1WorkflowStore({ db: env.WORKFLOW_DB })
defineWorkflowRuntime({ store, workflows })
createCloudflareWorkflowScheduledHandler({ runtime })
```

## Routes

- `POST /runs`: start a fulfillment workflow.
- `GET /runs`: list stored runs.
- `GET /runs/:runId`: read a run timeline.
- `POST /signals/payment`: deliver `payment-received`.
- `POST /sweep`: manually run a bounded sweep.
- Cron Trigger: runs the same bounded sweep once per minute.

## D1 setup

Create a D1 database:

```bash
pnpm dlx wrangler d1 create tanstack-workflow-d1-poc
```

Copy the returned `database_id` into `wrangler.toml`.

The D1 binding uses `migrations_dir = &quot;migrations&quot;`. The demo migration is a
copy of the package-owned artifact from:

```txt
packages/workflow-store-cloudflare-d1/migrations/0000_workflow_store.sql
```

Apply it locally:

```bash
pnpm migrate:local
```

Apply it remotely before deployment:

```bash
pnpm migrate:remote
```

Cloudflare documents D1 migrations as Wrangler-managed SQL files, and
`migrations_dir` can be configured per D1 binding.

## Local run

```bash
pnpm install
pnpm dev
```

Start a run:

```bash
curl -X POST &quot;http://localhost:8787/runs&quot; \
  -H &#39;content-type: application/json&#39; \
  -d &#39;{ &quot;orderId&quot;: &quot;d1-1&quot;, &quot;delayMs&quot;: 30000 }&#39;
```

Sweep after the delay:

```bash
curl -X POST &quot;http://localhost:8787/sweep&quot;
```

Deliver payment:

```bash
curl -X POST &quot;http://localhost:8787/signals/payment&quot; \
  -H &#39;content-type: application/json&#39; \
  -d &#39;{ &quot;runId&quot;: &quot;d1-fulfillment:d1-1&quot;, &quot;paymentId&quot;: &quot;pay-d1-1&quot; }&#39;
```

Inspect the timeline:

```bash
curl &quot;http://localhost:8787/runs/d1-fulfillment%3Ad1-1&quot;
```

## Deploy

```bash
pnpm migrate:remote
pnpm deploy
```

If `CRON_SECRET` is configured, `POST /sweep` requires:

```http
Authorization: Bearer &lt;CRON_SECRET&gt;
```

Cron Triggers still call the Worker `scheduled()` handler directly.
">
<input type="hidden" name="project[files][package.json]" value="{
  &quot;name&quot;: &quot;tanstack-workflow-cloudflare-d1-poc&quot;,
  &quot;private&quot;: true,
  &quot;type&quot;: &quot;module&quot;,
  &quot;scripts&quot;: {
    &quot;dev&quot;: &quot;wrangler dev --local&quot;,
    &quot;deploy&quot;: &quot;wrangler deploy&quot;,
    &quot;migrate:local&quot;: &quot;wrangler d1 migrations apply WORKFLOW_DB --local&quot;,
    &quot;migrate:remote&quot;: &quot;wrangler d1 migrations apply WORKFLOW_DB --remote&quot;,
    &quot;test:types&quot;: &quot;tsc&quot;,
    &quot;build&quot;: &quot;tsc --noEmit&quot;
  },
  &quot;dependencies&quot;: {
    &quot;@tanstack/workflow-cloudflare&quot;: &quot;https://pkg.pr.new/TanStack/workflow/@tanstack/workflow-cloudflare@0bc426bdf0be6415d1d4727f835419e3155a95a5&quot;,
    &quot;@tanstack/workflow-core&quot;: &quot;https://pkg.pr.new/TanStack/workflow/@tanstack/workflow-core@0bc426bdf0be6415d1d4727f835419e3155a95a5&quot;,
    &quot;@tanstack/workflow-runtime&quot;: &quot;https://pkg.pr.new/TanStack/workflow/@tanstack/workflow-runtime@0bc426bdf0be6415d1d4727f835419e3155a95a5&quot;,
    &quot;@tanstack/workflow-store-cloudflare-d1&quot;: &quot;https://pkg.pr.new/TanStack/workflow/@tanstack/workflow-store-cloudflare-d1@0bc426bdf0be6415d1d4727f835419e3155a95a5&quot;
  },
  &quot;devDependencies&quot;: {
    &quot;@cloudflare/workers-types&quot;: &quot;latest&quot;,
    &quot;typescript&quot;: &quot;6.0.3&quot;,
    &quot;wrangler&quot;: &quot;latest&quot;
  }
}">
<input type="hidden" name="project[files][tsconfig.json]" value="{
  &quot;compilerOptions&quot;: {
    &quot;lib&quot;: [&quot;ES2022&quot;, &quot;WebWorker&quot;],
    &quot;module&quot;: &quot;ESNext&quot;,
    &quot;moduleResolution&quot;: &quot;Bundler&quot;,
    &quot;noEmit&quot;: true,
    &quot;resolveJsonModule&quot;: true,
    &quot;skipLibCheck&quot;: true,
    &quot;strict&quot;: true,
    &quot;target&quot;: &quot;ES2022&quot;,
    &quot;types&quot;: [&quot;@cloudflare/workers-types&quot;]
  },
  &quot;include&quot;: [&quot;src&quot;]
}
">
<input type="hidden" name="project[files][wrangler.toml]" value="name = &quot;tanstack-workflow-cloudflare-d1-poc&quot;
main = &quot;src/worker.ts&quot;
compatibility_date = &quot;2026-05-30&quot;

[triggers]
crons = [&quot;* * * * *&quot;]

[[d1_databases]]
binding = &quot;WORKFLOW_DB&quot;
database_name = &quot;tanstack-workflow-d1-poc&quot;
database_id = &quot;00000000-0000-0000-0000-000000000000&quot;
migrations_dir = &quot;migrations&quot;
">
<input type="hidden" name="project[files][migrations/0000_workflow_store.sql]" value="create table if not exists &quot;workflow_schema_migrations&quot; (
  migration_id text primary key,
  package_name text not null,
  package_version text,
  applied_at integer not null
);

create table if not exists &quot;workflow_runs&quot; (
  run_id text primary key,
  workflow_id text not null,
  workflow_version text,
  status text not null,
  input text not null,
  output text,
  error text,
  awaiting text,
  waiting_for text,
  pending_approval text,
  wake_at integer,
  lease_owner text,
  lease_expires_at integer,
  created_at integer not null,
  updated_at integer not null
);

create index if not exists &quot;workflow_runs_status_idx&quot;
  on &quot;workflow_runs&quot; (status, updated_at);

create index if not exists &quot;workflow_runs_lease_idx&quot;
  on &quot;workflow_runs&quot; (status, lease_expires_at);

create table if not exists &quot;workflow_run_states&quot; (
  run_id text primary key,
  workflow_id text not null,
  workflow_version text,
  status text not null,
  input text not null,
  output text,
  error text,
  awaiting text,
  waiting_for text,
  pending_approval text,
  created_at integer not null,
  updated_at integer not null
);

create table if not exists &quot;workflow_event_locks&quot; (
  run_id text primary key,
  created_at integer not null
);

create table if not exists &quot;workflow_events&quot; (
  run_id text not null,
  event_index integer not null,
  event_type text not null,
  step_id text,
  event text not null,
  created_at integer not null,
  primary key (run_id, event_index)
);

create index if not exists &quot;workflow_events_type_idx&quot;
  on &quot;workflow_events&quot; (run_id, event_type);

create table if not exists &quot;workflow_timers&quot; (
  run_id text not null,
  signal_id text not null,
  workflow_id text not null,
  workflow_version text,
  wake_at integer not null,
  lease_owner text,
  lease_expires_at integer,
  primary key (run_id, signal_id)
);

create index if not exists &quot;workflow_timers_due_idx&quot;
  on &quot;workflow_timers&quot; (wake_at, lease_expires_at);

create table if not exists &quot;workflow_signal_deliveries&quot; (
  run_id text not null,
  signal_id text not null,
  created_at integer not null,
  primary key (run_id, signal_id)
);

create table if not exists &quot;workflow_schedules&quot; (
  schedule_id text primary key,
  workflow_id text not null,
  workflow_version text,
  schedule text not null,
  overlap_policy text not null,
  input text,
  next_fire_at integer,
  enabled integer not null,
  updated_at integer not null
);

create index if not exists &quot;workflow_schedules_due_idx&quot;
  on &quot;workflow_schedules&quot; (enabled, next_fire_at);

create table if not exists &quot;workflow_schedule_buckets&quot; (
  schedule_id text not null,
  bucket_id text not null,
  workflow_id text not null,
  workflow_version text,
  run_id text not null,
  fire_at integer not null,
  input text,
  overlap_policy text not null,
  status text not null,
  lease_owner text,
  lease_expires_at integer,
  started_at integer,
  primary key (schedule_id, bucket_id)
);

create index if not exists &quot;workflow_schedule_buckets_lease_idx&quot;
  on &quot;workflow_schedule_buckets&quot; (status, fire_at, lease_expires_at);

insert into &quot;workflow_schema_migrations&quot; (
  migration_id,
  package_name,
  package_version,
  applied_at
)
values (
  &#39;0000_workflow_store&#39;,
  &#39;@tanstack/workflow-store-cloudflare-d1&#39;,
  null,
  cast(unixepoch(&#39;subsec&#39;) * 1000 as integer)
)
on conflict (migration_id) do nothing;
">
<input type="hidden" name="project[files][src/runtime.ts]" value="import { defineWorkflowRuntime, every } from &#39;@tanstack/workflow-runtime&#39;
import { createCloudflareD1WorkflowStore } from &#39;@tanstack/workflow-store-cloudflare-d1&#39;
import { digestWorkflow, fulfillmentWorkflow } from &#39;./workflows&#39;

export interface Env {
  CRON_SECRET?: string
  WORKFLOW_DB: D1Database
}

export function createRuntime(env: Env) {
  const store = createCloudflareD1WorkflowStore({
    db: env.WORKFLOW_DB,
  })

  return defineWorkflowRuntime({
    store,
    workflows: {
      &#39;d1-fulfillment&#39;: {
        load: async () =&gt; fulfillmentWorkflow,
      },
      &#39;d1-digest&#39;: {
        load: async () =&gt; digestWorkflow,
        schedules: [
          {
            id: &#39;d1-digest-every-5m&#39;,
            schedule: every.minutes(5),
            overlapPolicy: &#39;skip&#39;,
            input: () =&gt; ({
              triggeredAt: Date.now(),
              scheduleId: &#39;d1-digest-every-5m&#39;,
            }),
          },
        ],
      },
    },
  })
}
">
<input type="hidden" name="project[files][src/worker.ts]" value="import { createCloudflareWorkflowScheduledHandler } from &#39;@tanstack/workflow-cloudflare&#39;
import { createRuntime, type Env } from &#39;./runtime&#39;
import type { FulfillmentInput, PaymentReceivedPayload } from &#39;./workflows&#39;
import type { WorkflowRuntimeDefinition } from &#39;@tanstack/workflow-runtime&#39;

interface StartRunBody {
  runId?: string
  orderId?: string
  amount?: number
  delayMs?: number
  readyAt?: number
}

interface PaymentSignalBody {
  runId: string
  paymentId?: string
  provider?: string
  signalId?: string
}

const scheduled = createCloudflareWorkflowScheduledHandler({
  runtime: ({ env }: { env: Env }) =&gt; createRuntime(env),
  maxScheduledRuns: 10,
  maxTimers: 25,
  maxDurationMs: 25_000,
  includeEvents: false,
})

export default {
  async fetch(request: Request, env: Env): Promise&lt;Response&gt; {
    const url = new URL(request.url)
    const runtime = createRuntime(env)

    try {
      if (url.pathname === &#39;/&#39; &amp;&amp; request.method === &#39;GET&#39;) {
        return json({
          ok: true,
          demo: &#39;tanstack-workflow-cloudflare-d1-poc&#39;,
          endpoints: {
            startRun: &#39;POST /runs&#39;,
            listRuns: &#39;GET /runs&#39;,
            getTimeline: &#39;GET /runs/:runId&#39;,
            paymentSignal: &#39;POST /signals/payment&#39;,
            sweep: &#39;POST /sweep&#39;,
          },
        })
      }

      if (url.pathname === &#39;/runs&#39; &amp;&amp; request.method === &#39;POST&#39;) {
        return json(
          await startRun(runtime, await readJson&lt;StartRunBody&gt;(request)),
        )
      }

      if (url.pathname === &#39;/runs&#39; &amp;&amp; request.method === &#39;GET&#39;) {
        return json(
          await runtime.store.listRuns({
            workflowId: url.searchParams.get(&#39;workflowId&#39;) ?? undefined,
            status:
              (url.searchParams.get(&#39;status&#39;) as
                | Parameters&lt;typeof runtime.store.listRuns&gt;[0][&#39;status&#39;]
                | null) ?? undefined,
            limit: Number(url.searchParams.get(&#39;limit&#39;) ?? 25),
            cursor: url.searchParams.get(&#39;cursor&#39;) ?? undefined,
          }),
        )
      }

      if (url.pathname.startsWith(&#39;/runs/&#39;) &amp;&amp; request.method === &#39;GET&#39;) {
        const runId = decodeURIComponent(url.pathname.slice(&#39;/runs/&#39;.length))
        const timeline = await runtime.store.getRunTimeline(runId)
        return timeline ? json(timeline) : json({ error: &#39;not-found&#39; }, 404)
      }

      if (url.pathname === &#39;/signals/payment&#39; &amp;&amp; request.method === &#39;POST&#39;) {
        return json(
          await deliverPayment(
            runtime,
            await readJson&lt;PaymentSignalBody&gt;(request),
          ),
        )
      }

      if (url.pathname === &#39;/sweep&#39; &amp;&amp; request.method === &#39;POST&#39;) {
        if (!isAuthorized(request, env))
          return json({ error: &#39;unauthorized&#39; }, 401)

        return json(
          await runtime.sweep({
            maxScheduledRuns: 10,
            maxTimers: 25,
            maxDurationMs: 25_000,
            leaseOwner: `manual:${Date.now()}`,
            includeEvents: false,
          }),
        )
      }

      return json({ error: &#39;not-found&#39; }, 404)
    } catch (error) {
      return json(
        { error: error instanceof Error ? error.message : &#39;Unknown error&#39; },
        400,
      )
    }
  },

  scheduled,
}

async function startRun(
  runtime: WorkflowRuntimeDefinition,
  body: StartRunBody,
) {
  const orderId = body.orderId ?? `order-${Date.now()}`
  const input: FulfillmentInput = {
    orderId,
    amount: body.amount ?? 4200,
    readyAt: body.readyAt ?? Date.now() + (body.delayMs ?? 30_000),
  }
  const runId = body.runId ?? `d1-fulfillment:${orderId}`

  return await runtime.startRun({
    workflowId: &#39;d1-fulfillment&#39;,
    runId,
    input,
    leaseOwner: &#39;http:start&#39;,
    includeEvents: true,
  })
}

async function deliverPayment(
  runtime: WorkflowRuntimeDefinition,
  body: PaymentSignalBody,
) {
  const paymentId = body.paymentId ?? `pay-${Date.now()}`
  const payload: PaymentReceivedPayload = {
    paymentId,
    provider: body.provider ?? &#39;demo&#39;,
  }

  return await runtime.deliverSignal({
    runId: body.runId,
    signalId: body.signalId ?? `payment:${paymentId}`,
    name: &#39;payment-received&#39;,
    payload,
    leaseOwner: &#39;http:payment&#39;,
    includeEvents: true,
  })
}

async function readJson&lt;T&gt;(request: Request): Promise&lt;T&gt; {
  return (await request.json()) as T
}

function isAuthorized(request: Request, env: Env) {
  if (!env.CRON_SECRET) return true
  return request.headers.get(&#39;authorization&#39;) === `Bearer ${env.CRON_SECRET}`
}

function json(body: unknown, status = 200) {
  return Response.json(body, { status })
}
">
<input type="hidden" name="project[files][src/workflows.ts]" value="import { createWorkflow } from &#39;@tanstack/workflow-core&#39;

export interface FulfillmentInput {
  orderId: string
  amount: number
  readyAt: number
}

export interface PaymentReceivedPayload {
  paymentId: string
  provider: string
}

export interface DigestInput {
  triggeredAt: number
  scheduleId: string
}

export const fulfillmentWorkflow = createWorkflow({
  id: &#39;d1-fulfillment&#39;,
  version: &#39;poc-v1&#39;,
}).handler(async (ctx) =&gt; {
  const input = ctx.input as FulfillmentInput

  const reservation = await ctx.step(&#39;reserve-inventory&#39;, (step) =&gt; ({
    reservationId: `res_${input.orderId}`,
    idempotencyKey: step.id,
    reservedAt: Date.now(),
  }))

  const firstSeenAt = await ctx.now()
  if (input.readyAt &gt; firstSeenAt) {
    await ctx.sleepUntil(input.readyAt)
  }

  const payment = await ctx.waitForEvent&lt;PaymentReceivedPayload&gt;(
    &#39;payment-received&#39;,
    {
      meta: {
        orderId: input.orderId,
        reservationId: reservation.reservationId,
      },
    },
  )

  const shipment = await ctx.step(&#39;ship-order&#39;, (step) =&gt; ({
    shipmentId: `ship_${input.orderId}`,
    idempotencyKey: step.id,
    shippedAt: Date.now(),
  }))

  return {
    orderId: input.orderId,
    amount: input.amount,
    reservation,
    payment,
    shipment,
  }
})

export const digestWorkflow = createWorkflow({
  id: &#39;d1-digest&#39;,
  version: &#39;poc-v1&#39;,
}).handler(async (ctx) =&gt; {
  const input = ctx.input as DigestInput

  const digest = await ctx.step(&#39;generate-digest&#39;, (step) =&gt; ({
    digestId: `digest_${input.scheduleId}_${input.triggeredAt}`,
    idempotencyKey: step.id,
    generatedAt: Date.now(),
  }))

  await ctx.step(&#39;publish-digest&#39;, (step) =&gt; ({
    digestId: digest.digestId,
    idempotencyKey: step.id,
    publishedAt: Date.now(),
  }))

  return digest
})
">
<input type="hidden" name="project[description]" value="generated by https://pkg.pr.new">
<input type="hidden" name="project[template]" value="node">
<input type="hidden" name="project[title]" value="tanstack-workflow-cloudflare-d1-poc">
</form>
<script>document.getElementById("mainForm").submit();</script>

</body></html>