Reqon

A declarative DSL for fetch, map, validate pipelines - built on Vague.

What is Reqon?

Reqon lets you define data synchronization pipelines in a readable, declarative language. Think of it like Temporal.io, but with a focus on API data fetching and transformation.

Example

mission SyncXeroInvoices {
  source Xero {
    auth: oauth2,
    base: "https://api.xero.com/api.xro/2.0"
  }

  store invoices: memory("invoices")
  store normalized: memory("normalized")

  action FetchInvoices {
    get "/Invoices" {
      paginate: offset(page, 100),
      until: length(response.Invoices) == 0
    }

    store response.Invoices -> invoices {
      key: .InvoiceID,
      partial: true
    }
  }

  action NormalizeInvoices {
    for invoice in invoices {
      map invoice -> StandardInvoice {
        id: .InvoiceID,
        amount: .Total,
        status: match .Status {
          "PAID" => "paid",
          "AUTHORISED" => "approved",
          _ => "pending"
        }
      }

      validate response {
        assume .amount >= 0
      }

      store response -> normalized { key: .id }
    }
  }

  run FetchInvoices then NormalizeInvoices
}

Installation

npm install reqon

Usage

CLI

reqon sync-invoices.vague --verbose
reqon sync-invoices.vague --auth ./credentials.json
reqon sync-invoices.vague --dry-run

Programmatic

import { execute } from 'reqon';

const source = `
  mission Test {
    source API { auth: bearer, base: "https://api.example.com" }
    store items: memory("items")
    action Fetch {
      get "/items"
      store response -> items { key: .id }
    }
    run Fetch
  }
`;

const result = await execute(source, {
  auth: { API: { type: 'bearer', token: 'your-token' } }
});

console.log(result.stores.get('items').list());

DSL Reference

Sources

Sources can be defined with explicit base URLs or by referencing an OpenAPI spec:

// Traditional: explicit base URL
source Name {
  auth: oauth2 | bearer | basic | api_key,
  base: "https://api.example.com"
}

// OAS-based: load from OpenAPI spec (base URL derived from spec)
source Petstore from "./petstore-openapi.yaml" {
  auth: bearer,
  validateResponses: true  // Optional: validate responses against OAS schema
}

Stores

store name: memory("collection")
store name: sql("table_name")
store name: nosql("collection")

Actions

action Name {
  // Steps: get/post/put/patch/delete, call, for, map, validate, store
}

HTTP Requests

Two styles are supported:

// Traditional: explicit HTTP method and path
get "/path" {
  paginate: offset(page, 100),
  until: response.items.length == 0,
  retry: { maxAttempts: 3, backoff: exponential }
}

// OAS-based: reference by Source.operationId
call Petstore.listPets {
  paginate: cursor(cursor, 20, "nextCursor"),
  until: response.pets.length == 0
}

When using OAS-based call, the HTTP method and path are resolved from the OpenAPI spec automatically.

Iteration

for item in collection where .status == "pending" {
  // nested steps
}

Mapping

map source -> TargetSchema {
  field: .sourceField,
  computed: .price * .quantity,
  status: match .state {
    "A" => "active",
    _ => "unknown"
  }
}

Validation

validate target {
  assume .amount > 0,
  assume .date >= .createdAt
}

Pipeline

// Sequential execution
run Step1 then Step2 then Step3

// Parallel execution with brackets
run [Step1, Step2] then Step3  // Step1 and Step2 run in parallel, then Step3

Durability Features

mission DurablePipeline {
  // Checkpoint after each step for resume-on-failure
  checkpoint: afterStep  // or onFailure

  // Enable time-travel debugging
  trace: full  // or minimal

  action WaitForApproval {
    // Resource-free pause (days/weeks without holding resources)
    pause {
      duration: "7d",
      resumeOn: timeout | webhook "/approved"
    }
  }

  run WaitForApproval
}

Durability guarantees

Run a mission as a durable execution (executionLog:) and an append-only event log lets an interrupted run — crash, deploy, kill -9 — resume exactly where it left off:

• Delivery: at-least-once (a step is never silently dropped).
• Effects: exactly-once where the API honours idempotency keys (mutating fetches carry a stable Idempotency-Key), otherwise at-least-once + keyed store dedup. Exactly-once on replay via recorded effect identity.
• Resume across restart: replay + fold; effects already applied are skipped.
• Backends: in-memory (tests), file (dev), SqliteExecutionLog (transactional, fsync-backed) for single-process self-hosting, and PostgresExecutionLog for multi-node.

These guarantees are proven by a crash-injection suite that kills the run at every event boundary and asserts no lost record and no duplicated effect (npm run test:crash). See DURABILITY.md for the full statement and the guarantee → test map.

OpenAPI Integration

Reqon can consume OpenAPI specs directly, eliminating the need for handwritten SDK code:

mission SyncPets {
  // Load API definition from OpenAPI spec
  source Petstore from "./petstore.yaml" {
    auth: bearer,
    validateResponses: true
  }

  store pets: memory("pets")

  action FetchPets {
    // Use operationId from spec - method and path are resolved automatically
    call Petstore.listPets

    store response.pets -> pets { key: .id }
  }

  run FetchPets
}

Benefits:

• No SDK required - The OpenAPI spec is the SDK
• Always up-to-date - Spec changes are picked up automatically
• Response validation - Validate API responses against the spec's schemas
• Auto-discovery - Base URLs, paths, and methods come from the spec

Development

npm run build      # Compile TypeScript
npm run test:run   # Run tests
npm run dev        # Watch mode

License

ISC