conveyor-graph-model

JSON documents, a type registry, and a loader for conveyor-graph. Separate repository / package so graph documents and vertex factories can version independently of the runtime.

Peer dependency: conveyor-graph. Optional: mysql2 (only if you use sql.query and open a real connection).

Document

Schema: schema/graph-config.json (also exported as GRAPH_CONFIG_SCHEMA).

{
  "id": "wp-posts",
  "vertices": [
    {
      "id": "query",
      "type": "sql.query",
      "config": {
        "sql": "SELECT * FROM wp_posts LIMIT 20",
        "connection": { "host": "127.0.0.1", "port": 13306, "database": "wordpress_db" }
      }
    },
    { "id": "print", "type": "log.print", "config": { "format": "json" } }
  ],
  "edges": [
    { "source": "query", "target": "print", "capacity": 16, "delivery": "required" }
  ]
}

when is a closed predicate DSL compiled at load time. Paths walk the StreamAgent (payload.kind, context.role). Equality is Object.is. Missing paths are false (except exists). Named { "pred": "isOnCall" } must be passed into loadGraph({ predicates }).

{
  "all": [
    { "path": "payload.kind", "equals": "record" },
    { "path": "payload.site", "in": ["north", "south"] }
  ]
}

Operators: equals, notEquals, exists, lt / lte / gt / gte, in, all, any, not, pred. Exactly one operator per node. field is accepted as an alias of path.

Loader

import { GraphAgent } from "conveyor-graph";
import { defaultRegistry, loadGraph } from "conveyor-graph-model";

const graph = loadGraph(doc, {
  registry: defaultRegistry(),
  predicates: { isOnCall: (agent) => agent.context.onCall === true },
});
const agent = new GraphAgent("session", {}, {}, graph);
await agent.send("query", "run-1", {});

loadGraph validates, initGraph(), defines each vertex from type, connects each edge, then seals.

Handlers that only exist in process (helper pipelines, ad-hoc functions) use type "instance". Pass them at load time:

loadGraph(doc, {
  registry: defaultRegistry(),
  instances: { classify: classifyHandler, sink: sinkHandler },
  // or a list zipped with doc.vertices
});

registry.bind("math.double", handler) both registers a factory that returns that function and remembers the mapping for export.

Export

import { exportGraph, exportRegistry } from "conveyor-graph-model";

const doc = exportGraph(graph, {
  registry,
  types: { "gateway.store": "sql.query" }, // optional overrides
  when: { "src-keep": { path: "payload.kind", equals: "keep" } },
});

Type recovery order: types[id], vertex.options.labels.type, registry.lookupType(handler), then "instance". Runtime builtins (graph/log, graph/skip, graph/error) are omitted unless includeBuiltins: true. Edge when functions cannot be reversed into the predicate DSL; pass when on export if you still have the documents.

exportRegistry(registry) returns { types } — factories stay in process.

Registry

registry.register("my.type", (spec) => async (agent) => {
  // spec.config is the vertex config object
  return agent.payload;
});

Unknown type fails closed.

Built-in types

| type | role | |---|---| | identity | pass payload through | | log.print | write a field (default payload) as JSON or inspect | | sql.query | run config.sql on context.db or config.connection |

Example

examples/mesh.json + examples/mesh.ts load a 15-vertex document (fan-out by kind, numeric band split, capacity-1 throttle into the sink), talk to three in-process TCP JSON-line servers via example-only socket.ask, stamp agent.context at every station, inject 36 items, and print each row at sink.

npx tsx examples/mesh.ts

stamp / socket.ask / collect live in the example host registry. Package builtins (identity, log.print, sql.query) stay in src/vertices.

Local layout

artifacts/conveyor-graph        runtime
artifacts/conveyor-graph-model  this package (file:../conveyor-graph)