@classytic/facts

Generic fact-projection kit — the domain-agnostic machinery behind rebuildable CQRS analytics projections (extracted from a production sales-analytics pipeline).

One factory call gives a domain everything except its own semantics:

| Machinery (this kernel owns) | Domain (the binding owns) | |---|---| | Generation manifest — active → rebuilding → catching_up → verified → active verified-cutover protocol, evidence stamps, cached reads, dual-write targets through every rebuild state, 3×TTL cutover time-guard | event → fact-row mappers (dedupe-key scheme, measures math, quality markers) | | projectFacts — the idempotent "small projection transaction": dedupe pre-read → ONE transaction: insert only-new facts + per-grain rollup $inc upserts | event subscriptions / relay wiring | | Rollup engine — '(none)' sentinel grain normalization, Decimal128 accumulators, deterministic rebuildRollup from facts in bounded civil-date chunks | the grain + measure list (declared in the spec) | | Backfill engine — keyset _id-asc iteration, per-(checkpointId, source) resumable checkpoints pinned to one generation, validated batch/maxDocs/sleep bounds, insert-dup counting | DocumentSource bindings (collection + filter + mapBatch) | | Reconcile framework — facts-side aggregation vs injected source-side aggregations, per-cell drift report | ReconcileSource aggregations | | Freshness — outbox diagnostics + manifest → {projectedThrough, lagMs, status} | the dedicated outbox store instance (host-owned, rule 16) | | Models + indexes (unique {generation, dedupeKey}, unique rollup grain), boot capability gate, syncIndexes | collection names, dimensions, secondaries (each with a rule-34 comment) |

import { createFactProjection } from '@classytic/facts';

const sales = createFactProjection<SalesFact>({
  name: 'sales',                      // manifest doc id + model namespace
  connection: mongoose.connection,
  collections: {
    facts: 'sales_fact',
    rollup: 'sales_daily',
    manifest: 'sales_projection_manifest',
    checkpoints: 'sales_projection_checkpoints',
  },
  civil: {                            // business-zone instant → civil labels
    dateOf: businessDateStr,          // e.g. Asia/Dhaka civil date
    weekOf: businessWeekOf,
    monthOf: businessMonthOf,
  },
  fact: {
    dimensions: {
      kind: { type: String, enum: ['ordered', 'fulfilled', 'returned', 'cancelled'], required: true },
      organizationId: { type: String, required: true },   // a DIMENSION here, not a scope
      skuRef: { type: String }, nodeRef: { type: String },
      channel: { type: String }, currency: { type: String },
      orderId: { type: String }, orderNumber: { type: String }, lineId: { type: String },
    },
    measures: [
      { name: 'orderedQty', nonNegative: true },
      { name: 'fulfilledQty', nonNegative: true },
      'grossRevenueMinor', 'discountMinor', 'taxMinor', 'netRevenueMinor', 'cogsMinor',
    ],
    markers: {
      qtySource: { values: ['event', 'reconstructed'], default: 'event' },
      costSource: { values: ['snapshot', 'missing'], default: 'missing' },
    },
    extraIndexes: [
      { fields: { generation: 1, organizationId: 1, civilDate: 1, skuRef: 1 },
        comment: 'branch dashboards / demand history: branch x day x SKU' },
    ],
  },
  rollup: {
    grain: ['organizationId', 'skuRef', 'nodeRef', 'channel', 'currency'],
    measures: ['orderedQty', 'fulfilledQty', 'grossRevenueMinor', 'discountMinor',
               'taxMinor', 'netRevenueMinor', 'cogsMinor'],
  },
  autoIndex: process.env.NODE_ENV !== 'production',   // rule 35
});

await sales.ensureReady();  // capability gate (transactions) + collection DDL

The correctness core (what the moved machinery guarantees)

• Idempotency — dedupe keys are deterministic BUSINESS-occurrence keys, never eventId; uniqueness is per (generation, dedupeKey). At-least-once redelivery, double envelope surfaces and backfill/live overlap all converge to one row.
• Fact↔rollup atomicity — projectFacts pre-reads existing dedupe keys (redelivery drives nothing, so the rollup can never double-$inc), then commits the only-new facts and their per-grain $inc upserts in ONE mongokit withTransaction. Every failure throws — the relay's retry/backoff/DLQ contract holds (a swallowed error would ack an unprojected row).
• Lossless cutover — the projector writes through manifest.getGenerationTargets(): [active] normally, [active, rebuild] in all three rebuild states, and cutover() refuses until the rebuild has been visible ≥ 3 × cache TTL, so every pod dual-writes before the flip.
• Deterministic rebuild — rebuildRollup re-aggregates facts through the repository AggRequest IR in bounded chunks; sentinel normalization + the civil consistency contract keep rebuilt rows byte-equal to live rows.

New domain checklist (a binding lands in ~300 lines)

To stand up purchases-analytics (or POS-shift, inventory-movement, support…):

1. Spec (~60 lines) — createFactProjection({ name: 'purchases', ... }): collections, dimensions, integer measures (nonNegative for quantities), marker vocabulary, rollup grain, secondaries (rule-34 comments mandatory).
2. Civil port (~5 lines) — bind dateOf/weekOf/monthOf over the host's business-date helpers. Contract: weekOf(d) === isoWeekOfCivilDate(dateOf(d)) and monthOf(d) === dateOf(d).slice(0, 7) (rebuild equivalence depends on it).
3. Event mappers (~100 lines) — pure event → FactRow[] per event type: build the dedupe key (placed:{docNo}:{lineId}-style), use engine.civilFields(occurredAt) for the time block, assertValidQty / safeMulMinor for every measure, stamp quality markers. The subscribing handler resolves generations implicitly — just call engine.projectFacts(rows) (dual-write is automatic).
4. DocumentSource bindings (~80 lines) — per durable source collection: { name, filter, read: (q, n) => coll.find(q).sort({_id:1}).limit(n).toArray(), estimateTotal, mapBatch }. mapBatch reuses the same dedupe/marker rules as the live mappers so a backfilled generation is row-equivalent. Docs must have ObjectId _ids (the checkpoint cursor type).
5. ReconcileSource aggregations (~50 lines) — independent per-cell totals from the source docs for the measures each source is authoritative for.
6. Host wiring — dedicated outbox store + relay (durability is host-owned, rule 16), a strict single-consumer transport lane, subscriptions, ops resource calling manifest.*, runBackfill, rebuildRollup, reconcile, getFreshness(store).

Rebuild/cutover, checkpointed backfill, drift reports, freshness and all indexes come for free.

Notes & sharp edges

• No tenant plugin — deliberate. These are analytical projections: an org/branch id among the dimensions is a queryable dimension, not a scope. Scope enforcement belongs to the query surface (lens semantic models / resource permissions). A tenant plugin here would break projector system writes (no request scope) and HQ rollup reads.
• allowNonTransactional is the standalone-Mongo dev fallback only. In fallback mode a duplicate-only insert race rolls up exactly the subset that inserted (subtractFailedInserts) — the invariant holds; production leaves it false and ensureReady() fails closed without transactions.
• Schema backstop is loud — insertMany runs with throwOnValidationError: true; a row failing the safe-integer/non-negative validators aborts the transaction (or backfill batch) instead of being silently skipped while the rollup counts it.
• Mapper identity is fail-fast — two rows with the same dedupeKey in one input batch are rejected before a transaction starts; the kernel never silently chooses between conflicting representations of one occurrence.
• Rollup measures are Decimal128 — never serialize raw; use rollupMeasureToNumber (throws on non-integer / unsafe magnitudes).
• Freshness port ≠ primitives OutboxStore — freshness needs two indexed diagnostic reads (oldestPendingAgeMs, newestDeliveredInstant) the relay contract deliberately does not carry; bind them over your store. Import OutboxStore / EventTransport contracts directly from @classytic/primitives subpaths (P4) — this package never re-exports them.
• Model collisions throw (rule 21) — two engines of one projection need two connections, or forceRecreate: true in hot-reload/test fixtures.
• Peers: @classytic/mongokit ≥3.25, @classytic/repo-core ≥0.14, @classytic/primitives ≥0.14, mongoose ≥9.4.1, zod ≥4.

Testing

npm run test:unit          # pure machinery (guards, rollup grouping, civil, spec)
npm run test:integration   # MongoMemoryReplSet: manifest FSM, §4.3 txn, rebuild,
                           # backfill resume, reconcile, freshness