@x12i/countex-pipelines

@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