hazo_collect

Collector-manager engine for the Ocdata platform. Handles plugin discovery, contract validation, execution orchestration, result persistence, and watermark/health tracking.

Installation

npm install hazo_collect

Peer deps: hazo_core, hazo_connect. Optional: hazo_secure (for secrets injection).

Module formats

hazo_collect ships as ESM, but every subpath (., ./server, ./sdk) is resolvable from a CommonJS host as well — including a Next.js project ("type": "commonjs") whose worker runs under tsx. Both import('hazo_collect/server') and require('hazo_collect/server') resolve, and a plugin that registers via import { defineCollector } from 'hazo_collect/sdk' is discovered by a discover() call made from the CJS worker — the registry is shared process-wide regardless of how each side is loaded.

Quick start

1. Define a collector (SDK)

import { defineCollector } from 'hazo_collect/sdk';
import { parseManifest } from 'hazo_collect';

defineCollector({
  manifest: parseManifest({
    name: 'my_source',
    kind: 'source',
    version: '1.0.0',
    runtime: 'node',
    entry: './my_source.js',
    idempotency_key: ['record_id'],
  }),
  async run(ctx) {
    const rows = [{ record_id: 'r1', data: 'hello' }];
    await ctx.write({ plugin: 'my_source', rows, idempotency_key: ['record_id'] });
    return { records_fetched: rows.length, records_written: rows.length };
  },
});

2. Create a manager and run the collector

import { createManager, nodeRuntime, createInMemoryRegistry } from 'hazo_collect/server';

const manager = createManager({
  getHazoConnect: () => adapter,      // hazo_connect adapter
  getCollector: name => registry.get(name),
  runtime: nodeRuntime,
});

const result = await manager.runNow({ plugin: 'my_source' });
// result.data: RunResult — status, records_fetched, records_written, errors, watermark

API

hazo_collect (client-safe)

| Export | Description | |---|---| | CONTRACT_VERSION | Current contract version string ("1.0.0") | | parseManifest(obj) | Parse + validate a plugin manifest; throws HazoValidationError on invalid input | | parseRunResult(obj) | Parse + validate a RunResult object | | parseRunEnvelope(obj) | Parse + validate a RunEnvelope; applies defaults | | validateInputs(manifest, payload) | Validate run inputs against a manifest's JSON schema |

hazo_collect/server

| Export | Description | |---|---| | createManager(opts) | Main orchestration pipeline — resolve → validate → run → persist | | nodeRuntime | Execution adapter for in-process Node.js collectors | | createInMemoryRegistry() | In-memory registry backed by the SDK Map | | persistRegistry(adapter, manifests) | Upsert manifests into hazo_collect_plugin_registry; quarantines invalid entries | | discover(opts) | Scan directories for manifest.json files and return parsed entries | | createWriteAdapter(adapter, runId) | Create a ctx.write function bound to a specific run |

hazo_collect/sdk

| Export | Description | |---|---| | defineCollector(def) | Register a collector in the process-global registry | | getCollector(name) | Look up a registered collector by name | | listCollectors() | Return all registered collectors | | defineSink(def) | Register a sink (data destination) | | getSink(name) | Look up a registered sink | | resetCollectorRegistry() | Clear all registrations (test hygiene) |

The registry is a process-global singleton keyed on globalThis[Symbol.for('hazo_collect.collectors')]. This guarantees that a collector registered through hazo_collect/sdk is visible to discover() / listCollectors() in hazo_collect/server, even when the two entry points are bundled separately or loaded under different module formats (see Module formats below).

defineCollector / defineSink accept the manifest input shape (ManifestInput): fields with defaults — timezone, timeout_sec, concurrency, retry.* — may be omitted. The registry runs parseManifest to fill defaults before the runtime reads them, so wrapping the literal in parseManifest(...) (as in the quick start above) is optional.

Database tables

Run migrations from hazo_collect/ddl/sqlite.sql (SQLite) or hazo_collect/ddl/postgres.sql (PostgreSQL):

• hazo_collect_plugin_registry — discovered manifest catalogue with validity + quarantine info
• hazo_collect_plugin_runs — per-run scorecard (status, records, contract_version, watermark)
• hazo_collect_plugin_health — rolling health state (consecutive failures, last status)
• hazo_collect_landing — idempotent landing zone for collector output rows

Tailwind v4 (@source required)

@source "../node_modules/hazo_collect/dist";

License

MIT