DataFor eachPostgreSQL

Import thousands of customers with durable per-row progress

Validate a dataset once, process rows concurrently with separate retry identity, upsert valid customers, and return an ordered import report.

A JavaScript loop can transform an array, but it cannot give every row its own durable attempt, progress, cancellation point, and bounded concurrency. <for-each> can.

The workflow at a glance

Text

Validated import webhook
  → Prepare rows
  → For each customer, concurrency 16
      → Normalize
      → Valid? ─ yes → Upsert PostgreSQL
               └ no  → Record skipped result
  → Summarize ordered results

The workflow

WOML

<woml>
  <workflow
    id="customer-import"
    name="Customer import"
    description="Normalize and persist a batch with durable row-level progress."
    version="1.0.0"
  >
    <config concurrency="2" timeout="30m" queue="imports" />

    <triggers>
      <webhook
        id="importCustomers"
        path="/webhooks/customer-imports"
        method="POST"
        auth="bearer"
        secret="{{secrets.IMPORT_WEBHOOK_TOKEN}}"
      >
        <schema>
          {
            "type": "object",
            "required": ["importId", "customers"],
            "properties": {
              "importId": { "type": "string" },
              "customers": {
                "type": "array",
                "maxItems": 10000,
                "items": {
                  "type": "object",
                  "required": ["externalId", "email", "name"],
                  "properties": {
                    "externalId": { "type": "string" },
                    "email": { "type": "string" },
                    "name": { "type": "string" }
                  }
                }
              }
            }
          }
        </schema>
      </webhook>
    </triggers>

    <steps>
      <step id="prepareImport" name="Prepare import">
        <script>
          return {
            importId: context.payload.importId,
            customers: context.payload.customers
          };
        </script>
      </step>

      <for-each
        id="processCustomers"
        name="Process customers"
        items="{{context.steps.prepareImport.customers}}"
        concurrency="16"
      >
        <step id="normalizeCustomer" name="Normalize customer">
          <script>
            const email = context.item.email.trim().toLowerCase();
            const name = context.item.name.trim().replace(/\s+/g, " ");

            return {
              externalId: context.item.externalId,
              email,
              name,
              valid: /^[^@\s]+@[^@\s]+\.[^@\s]+$/.test(email) && name.length > 0,
              row: context.iteration.index
            };
          </script>
        </step>

        <choose id="persistCustomer" name="Persist valid customer">
          <when test="{{context.steps.normalizeCustomer.valid}}">
            <step id="upsertCustomer" name="Upsert customer" retry="3" retry-backoff="exponential">
              <script>
                const db = services.db({
                  driver: "postgres",
                  connection: secrets.POSTGRES_URL
                });

                await db.execute({
                  text: `
                    INSERT INTO customers (external_id, email, name)
                    VALUES ($1, $2, $3)
                    ON CONFLICT (external_id)
                    DO UPDATE SET email = EXCLUDED.email, name = EXCLUDED.name
                  `,
                  values: [
                    context.steps.normalizeCustomer.externalId,
                    context.steps.normalizeCustomer.email,
                    context.steps.normalizeCustomer.name
                  ]
                }, { name: "upsert-imported-customer" });

                return {
                  row: context.steps.normalizeCustomer.row,
                  externalId: context.steps.normalizeCustomer.externalId,
                  status: "imported"
                };
              </script>
            </step>
            <result value="{{context.steps.upsertCustomer}}" />
          </when>

          <otherwise>
            <step id="skipCustomer" name="Skip invalid customer">
              <script>
                return {
                  row: context.steps.normalizeCustomer.row,
                  externalId: context.steps.normalizeCustomer.externalId,
                  status: "invalid"
                };
              </script>
            </step>
            <result value="{{context.steps.skipCustomer}}" />
          </otherwise>
        </choose>

        <result value="{{context.steps.persistCustomer}}" />
      </for-each>

      <step id="summary" name="Build import summary">
        <script>
          const results = context.steps.processCustomers.results;
          return {
            importId: context.steps.prepareImport.importId,
            total: context.steps.processCustomers.total,
            imported: results.filter((row) => row.status === "imported").length,
            invalid: results.filter((row) => row.status === "invalid").length,
            results
          };
        </script>
      </step>
    </steps>
  </workflow>
</woml>

Run and test it

Create the customers table with a unique external_id, then configure and run WOML:

Terminal

woml secrets set IMPORT_WEBHOOK_TOKEN
woml secrets set POSTGRES_URL
woml check customer-import.woml
woml run customer-import.woml

Send a batch with the authenticated curl command printed by the runtime. Start with a few rows, then increase the batch while watching item progress in woml inspect.

Why this shows WOML's range

The markup exposes data ingress, bounded fan-out, per-row decisions, and aggregation. JavaScript handles normalization, PostgreSQL handles application data, and Rust owns item identity, attempts, recovery, progress, and cancellation. The result remains understandable when the import grows from ten rows to ten thousand.