jusbid-event-bus — Redis Streams Activity Logging

Production-oriented, provider-agnostic event bus. Business microservices depend only on publish(); the activity-log-consumer is a separate background process that batches events into MongoDB.

→ For "how do I use this in my service", see HOW_TO_USE_IN_MICROSERVICES.md. That doc walks through exactly what a business service sends, what (if anything) it gets back, and what happens when things fail downstream.

Files created

shared/
  logger/index.js                        structured JSON logger (stand-in for repo's existing logger)

  redis-client/index.js                  ★ your actual @jusbid/redis package (uploaded, lightly
                                          adapted — see "Connection reuse" below). Owns the ONE
                                          Redis connection every service — and the event bus — shares.

  event-bus/
    index.js                             ★ ENTRYPOINT — business services import only this
    topics.js                            ★ canonical topic-name registry (single source of truth)
    interface/EventBus.js                abstract publish/consume/close contract
    config/index.js                      builds config.topics.<TOPIC_KEY> for every registered topic
    config/topicConfig.js                generic env-var-prefix → topic-config builder
    errors/EventBusError.js              InvalidEventError / PublishError / PoisonMessageError
    utils/validateEvent.js               structural validation before publish
    utils/BatchBuffer.js                 generic size-or-time flush trigger
    utils/withTimeout.js                 bounds publish() against a hung/unreachable Redis
    utils/mongoBatchPersist.js           reusable Mongo insertMany + dup-key/retry classification
    utils/metrics.js                     in-memory counters (swap for prom-client etc. later)
    redis/RedisProducer.js               xAdd wrapper (via the shared client, with a publish timeout)
    redis/RedisConsumer.js               xReadGroupNew/xAck/xPending/xAutoClaim, batching, DLQ, shutdown
    redis/RedisEventBus.js               ties producer+consumer together as the EventBus impl
    __tests__/                           28 tests, see "Testing" below

services/
  activity-log-consumer/
    consumer.js                          standalone process entrypoint (Mongo connect + graceful shutdown)
    batchHandler.js                      thin wrapper around utils/mongoBatchPersist.js
    models/ActivityLog.js                Mongoose schema + indexing rationale
    __tests__/batchHandler.test.js

  property-service-example/
    propertyService.example.js           shows a business service calling eventBus.publish()

.env.example                             every env var this package reads, with defaults
package.json / jest config               workspace-level (delete/merge into your monorepo's own)

Multiple log-type microservices (error logs, history logs, OTP, etc.): topics.js + config/topicConfig.js already generalize the config layer so any number of topics work without new config-parsing code — see config/index.js, which currently registers ACTIVITY_LOG, ERROR_LOG, HISTORY_LOG, and OTP_SEND. error-log-consumer and history-log-consumer services follow the exact same 3-file shape as activity-log-consumer (a Mongoose model + a toDocument mapper passed into persistBatchToMongo + a consumer.js entrypoint reading config.topics.ERROR_LOG/.HISTORY_LOG) — not included in this deliverable to keep it focused, but they're a copy-paste-and-adjust- fields exercise from activity-log-consumer, not new design work. OTP is different — it shouldn't batch (OTP_SEND's defaults are already batchSize:1, batchWaitMs:50, see config/index.js), its consumer calls an SMS/email provider instead of Model.insertMany, and for security the OTP code itself should never be put on the event — publish a request ("send an OTP to user X"), let the consumer generate and store the code. Ask if you want this one built out too.

This was built standalone (no existing repository was provided to inspect) — see Assumptions below for what would need reconciling against your real codebase.

Architecture

Microservices → eventBus.publish("activity-log", event)   [XADD, fire-and-forget]
                        │
                        ▼
                  Redis Stream: activity-log
                  Consumer Group: activity-log-writers
                        │
                        ▼
              activity-log-consumer (background process)
              XREADGROUP → BatchBuffer (500 events OR 2s)
                        │
                        ▼
              MongoDB.insertMany(batch, {ordered:false})
                        │
              ┌─────────┴─────────┐
              ▼                   ▼
        success/duplicate      real failure
              │                   │
            XACK              left pending
                                   │
                          reclaimed via XCLAIM after
                          claimIdleMs, retried up to
                          maxRetries, then → DLQ + XACK

Connection reuse (no per-microservice/per-connection duplication)

shared/event-bus does not open its own Redis connection. It requires shared/redis-client — your actual @jusbid/redis package, uploaded and adapted here (see "Assumptions" for the one-line change) — the same way every other part of a microservice would (caching, rate limiting, etc). Because Node caches require() results per resolved path, every module in the process that does require('@jusbid/redis') — including RedisEventBus — gets the exact same singleton client instance. There is one lazily-established connection per process, full stop; nothing in this package calls createClient()/connect() a second time.

One caveat worth knowing about, discovered while wiring this up: @jusbid/redis's reconnectStrategy retries indefinitely with growing backoff on a permanently-unreachable Redis, and its underlying .connect() promise never rejects on its own in that case — it just keeps retrying forever. That's a reasonable default for a long-running background consumer (it should wait for Redis to come back), but it's dangerous for publish(), which usually runs inline in a request path — an unbounded await there would hang the caller's request forever instead of degrading gracefully, undermining requirement #13. RedisProducer guards against this explicitly with utils/withTimeout.js: publish() is bounded to EVENT_BUS_PUBLISH_TIMEOUT_MS (default 5000ms) and always returns (null on timeout/failure) within that window, regardless of how long the shared client keeps retrying underneath.

Environment variables

See .env.example for the full, authoritative list with defaults. Highlights:

| Variable | Default | Purpose | |---|---|---| | EVENT_BUS_PROVIDER | redis | redis now; kafka once implemented — business code unaffected either way | | REDIS_URL (or REDIS_HOST/PORT/USERNAME/PASSWORD/TLS) | redis://localhost:6379 | read by shared/redis-client, not by the event bus directly | | EVENT_BUS_PUBLISH_TIMEOUT_MS | 5000 | max time publish() will wait before giving up — see "Connection reuse" above | | <TOPIC>_STREAM / _CONSUMER_GROUP / _CONSUMER_NAME / _DLQ_STREAM | per-topic, see .env.example | independently configurable per topic (ACTIVITY_LOG, ERROR_LOG, HISTORY_LOG, OTP_SEND) | | <TOPIC>_BATCH_SIZE | 500 (log topics), 1 (OTP) | flush trigger #1 | | <TOPIC>_BATCH_WAIT_MS | 2000 (log topics), 50 (OTP) | flush trigger #2 | | <TOPIC>_MAX_RETRIES | 5 (log topics), 3 (OTP) | deliveries allowed before DLQ | | <TOPIC>_CLAIM_IDLE_MS | 30000 (log topics), 10000 (OTP) | how long a message may sit unacked before being reclaimed — must stay well above _BATCH_WAIT_MS, see note below |

Operational note on CLAIM_IDLE_MS vs BATCH_WAIT_MS: a message sitting in a healthy consumer's local batch buffer, waiting out its own batchWaitMs window, is still technically "pending/unacked" in Redis. If claimIdleMs were set close to (or below) batchWaitMs, another consumer's reclaim loop could steal and reprocess that message before the original consumer got a chance to ack it — wasteful, though harmless thanks to the idempotent event_id unique index. The defaults (30s vs 2s, a 15x margin) avoid this in practice; keep a similar margin if you tune these.

Usage example (business microservice)

const { getEventBus } = require('../../shared/event-bus');
const eventBus = getEventBus();

await eventBus.publish('activity-log', {
  event_id: crypto.randomUUID(),
  action: 'PROPERTY_CREATED',
  service: 'property-service',
  tenant_id: actor.tenantId,
  user_id: actor.userId,
  entity_type: 'property',
  entity_id: property._id,
  metadata: { name: property.name },
  created_at: new Date(),
});

Full walkthrough (including what comes back and failure-mode behavior): HOW_TO_USE_IN_MICROSERVICES.md.

Consumer startup

cp .env.example .env   # adjust REDIS_URL / MONGO_URL for your environment
node services/activity-log-consumer/consumer.js

Run multiple replicas by giving each a distinct ACTIVITY_LOG_CONSUMER_NAME — Redis Streams consumer groups split the stream across them automatically.

Retry / DLQ behavior

1. A batch handler classifies every message as ackIds (persisted, or a benign duplicate event_id) or retryIds (genuine failure — e.g. Mongo unreachable).
2. ackIds are XACK'd immediately. retryIds are left pending.
3. A background reclaim loop periodically checks for messages idle longer than <TOPIC>_CLAIM_IDLE_MS, using the shared client's xPending (to discover idle ids and their delivery counts — only XPENDING's extended form exposes delivery count) followed by xAutoClaim (to physically transfer ownership so this consumer can act on them):
   • delivery count ≤ <TOPIC>_MAX_RETRIES → reclaimed and retried
   • delivery count > <TOPIC>_MAX_RETRIES → published to the topic's DLQ stream with the original payload + failure reason, then XACK'd on the original stream so it stops blocking further reclaim cycles.
4. Malformed (non-JSON) payloads skip straight to the DLQ — they can never succeed no matter how many times they're retried.

Failure-mode tradeoff (requirement #13)

publish() never throws for infrastructure failures (Redis down, etc.) — it logs internally and returns null. A business transaction (e.g. "property created") is a completed fact regardless of whether its activity-log entry made it onto the stream. Rolling back or blocking the business write because the logging pipeline is unavailable would be strictly worse for users than accepting a rare gap in the audit trail.

Testing

npm install
npm test

28 tests across 5 suites, all using either the real EventBus classes directly or a hand-rolled FakeStreamRedis in-memory double shaped like the shared @jusbid/redis client's own Streams exports (xAdd, xGroupCreate, xReadGroupNew, xAck, xPending, xAutoClaim) — chosen over ioredis-mock because that library implements a different Redis client's API and doesn't support Streams commands at all; see shared/event-bus/__tests__/helpers/FakeStreamRedis.js.

Covered: publish success/validation/infra-failure/publish-timeout against a hung connection, consumer-group creation idempotency (BUSYGROUP handling), size-triggered flush, time-triggered flush, successful batch processing, a Mongo-failure → reclaim → successful- retry cycle (deterministic, via a controllable fake clock rather than racing real timers), DLQ routing after maxRetries is exceeded, two consumers splitting a stream's workload without double-processing, graceful shutdown flushing the in-flight batch, and (at the Mongo- handler level) duplicate-event_id idempotent acknowledgement vs genuine write-error retry classification.

Not covered by automated tests (would need a real Redis + Mongo, or mongodb-memory-server/testcontainers, neither of which had network access in this sandbox): consumer-restart-after-crash against a real Redis server, and end-to-end load/throughput behavior. The reclaim-loop tests exercise the same code path a restart-after-crash would hit (pending entries idle past claimIdleMs), so the logic is covered even though the "kill -9 the process" scenario itself isn't simulated.

Assumptions and decisions made

• shared/redis-client/index.js is your actual uploaded @jusbid/redis package, with exactly one line changed: it imports { createLogger } from this deliverable's shared/logger and builds a logger from it, instead of { logger } from @jusbid/logger (which doesn't exist in this standalone sandbox). In your real monorepo, delete those two substitute lines and restore the original const { logger } = require('@jusbid/logger'); — nothing else in the file was touched, and nothing in shared/event-bus needed to change for that swap since it only depends on the exported function shapes (xAdd, xGroupCreate, xReadGroupNew, xAck, xPending, xAutoClaim, redisClient.ping/.quit), not on how the file gets its logger.
• shared/logger/index.js — a minimal structured-JSON logger, since @jusbid/logger itself wasn't provided. If you have it, point shared/redis-client's one substitute line back at it and delete shared/logger entirely.
• Mongoose connection (services/activity-log-consumer/consumer.js) — connects directly via mongoose.connect(MONGO_URL). If your repo has a shared DB-connection module with pooling/options already configured, use that instead of a bare connect() call.
• Package name @jusbid/event-bus — used in doc examples per your spec's naming; the actual files live under shared/event-bus/ and are required via relative paths in this standalone deliverable. Wire up an actual workspace package (package.json with that name) if your monorepo uses npm/yarn workspaces for internal packages.
• Kafka implementation — intentionally NOT built (out of scope per the spec — "do not attempt exactly-once", Kafka wasn't requested to be implemented, only made swappable). shared/event-bus/index.js documents exactly where a KafkaEventBus would plug in.
• error-log-consumer / history-log-consumer / otp-consumer — the config layer (topics.js, config/topicConfig.js) and the reusable Mongo-persist logic (utils/mongoBatchPersist.js) are built and tested; the actual service folders for these three aren't included here to keep this deliverable focused on the connection-reuse change you asked for. activity-log-consumer is the template — same 3 files, different Mongoose schema + toDocument mapping. Ask if you want these built out.