@x12i/countex-pipelines

Generic pre-aggregated processing counters for records moving through pipelines.

Countex-pipelines tracks status transitions (in_progress → done | failed) and serves rollups by subject, pipeline, and dimension — sub-second when the hot layer is warm, with no source-document scans at query time.

Package policy: Generic only. No host-product types, client names, or embedded scan logic. The host supplies historical replay via CountexRebuildSource when needed.

Default deployment: single machine, single process

The simplest and recommended starting config needs only MongoDB — no Redis required.

export MONGO_URI="mongodb://127.0.0.1:27017"
# REDIS_URL is optional — omit it for single-machine mode

import { createCountexFromEnv } from "@x12i/countex-pipelines";

const client = await createCountexFromEnv();
await client.initialize({ tenantId: "tenant-1", fromCounterStore: true });

await client.ingest({
  tenantId: "tenant-1",
  subject: "entities",
  recordId: "rec-42",
  pipelineId: "enrich",
  status: "done",
  occurredAt: new Date().toISOString(),
});

const rollup = await client.rollup({ tenantId: "tenant-1" });
console.log(rollup.bySubject, rollup.byPipeline);

await client.close();

When REDIS_URL is absent, countex-pipelines uses an in-process MemoryHotStore. Mongo remains the source of truth; on restart the hot layer is empty and is re-warmed from Mongo automatically on initialize or first query. Do not run multiple processes with MemoryHotStore — use Redis for multi-server.

| Config | Hot layer | When to use | |--------|-----------|-------------| | MONGO_URI only | In-process memory | One machine, one process (default) | | MONGO_URI + REDIS_URL | Shared Redis | Multiple servers, same counters | | COUNTEX_READ_THROUGH=1 | None (read Mongo) | Minimal infra, low volume |

Three consumption surfaces (choose one)

1. Primitives (library-first)

Import pure functions and wire storage yourself:

import {
  classifyRecord,
  classifyPipeline,
  buildDedupKey,
  MemoryHotStore,
  MongoDurableStore,
  PipelinesEngine,
} from "@x12i/countex-pipelines";
// or: import from "@x12i/countex-pipelines/primitives"

2. Embedded facade (in-process)

import { createCountexFromEnv } from "@x12i/countex-pipelines";
// or: import from "@x12i/countex-pipelines/client"

3. HTTP API (optional separate package)

Run as a standalone service for remote or non-Node consumers:

npm install @x12i/countex-pipelines-server
MONGO_URI=... countex-pipelines-server

See @x12i/countex-pipelines-server.

Ingest is write-through to Mongo — HTTP 200 means durably recorded.

Counter model

| Field | Role | |-------|------| | tenantId | Isolation scope | | subject | Host-defined subject type | | recordId | Item being processed | | pipelineId | Host-defined processor / workflow id | | dimensions | Optional tags for rollup axes | | Metric | succeeded | pending | failed | running (derived) |

Classification (per record)

| Bucket | Rule | |--------|------| | succeeded | At least one pipeline done, and no pipeline failed | | failed | At least one pipeline failed (wins over done) | | pending | No terminal status yet | | running | Per-pipeline: latest status is in_progress |

Starting from the middle

1. Redis/memory empty, Mongo has counters → initialize({ tenantId, fromCounterStore: true }) reloads and warms the hot layer.
2. Counters never existed, host has history → implement CountexRebuildSource.scan() and call initialize({ tenantId, fromSource }) (in-process only).
3. Process restart → same as (1); Mongo is source of truth.

Install

npm install @x12i/countex-pipelines mongodb
# optional for multi-server:
npm install ioredis

Node 18+.

Docs

• Use-case playbook

License

MIT