NatsStore

NatsStore Reference for NatsStore — the ORM's NoSQL store backed by NATS and its JetStream key/value layer. Source: src/nosql/nats/index.ts · Tests: tests/nosql/nats.test.ts Overview NATS is a lightweight publish/subscribe and request/reply messaging system. Its JetStream layer adds persistence, including a key/value (KV) store built on top of streams. NATS has no query language and does not fit the SQL-shaped Dialect interface, so NatsStore implements the minimal NoSqlStore marker interface (connect/disconnect/isConnected/getClient) and exposes NATS's real operations — core pub/sub, request/reply, and JetStream KV — instead of forcing everything into a query(sql) shape. Key facts: Identity: name = 'nats', library = 'nats' (both readonly). Driver: uses the nats driver (nats.js). It is not a hard dependency — the driver is lazy-loaded inside connect() via require('nats'), so importing this module never pulls in the driver unless a real connection is actually opened. Injected client: a pre-built NatsConnection can be supplied through the client option. When present, connect() adopts it directly and never calls require('nats'). This is how tests (and callers managing their own connection) run without the real driver or a network. Encoding rules NATS message payloads are byte arrays (Uint8Array). The payload type accepted by publish(), request(), and kvPut() is: A string is UTF-8 encoded via TextEncoder. A Uint8Array is passed through unchanged. On the read side, subscribe() handlers and request() replies deliver the driver's raw message; kvGet() returns the entry's raw Uint8Array value (or null). Decode with TextDecoder (or the real driver message's own .string() / .json() helpers) as needed. Connection Real connection connect() lazy-loads the driver and calls nats.connect({ servers, token, user, pass }). All four option fields are passed straight through to the driver. If the driver is not installed or the connection fails, connect() throws a ConnectionError ("Unable to connect to NATS: ...") rather than a raw module-not-found error. Injected client connect() is idempotent: calling it again while already connected returns immediately. Options Methods Lifecycle connect(): Promise<void> Opens (or adopts) the connection. Idempotent when already connected. With a client option it adopts the injected connection; otherwise it lazy-loads nats and connects using the auth options. On failure it resets internal state and throws ConnectionError. disconnect(): Promise<void> Tears down the connection. Prefers a graceful drain() (flushes pending messages and unsubscribes); if the connection has no drain, it falls back to close(). Secondary teardown failures are swallowed since the connection is discarded regardless. Clears the KV bucket cache and marks the store disconnected. Safe to call multiple times (idempotent). isConnected(): boolean Returns true only when connected and a connection object is held. getClient(): NatsConnectionLike Returns the underlying nats connection for anything not wrapped here. Throws ConnectionError ("Not connected to NATS") if there is no active connection. Core pub/sub + request/reply publish(subject: string, data: NatsPayload): void Publishes data to subject, encoding the payload per the encoding rules. Synchronous (returns void). Requires an active connection — throws ConnectionError otherwise. Driver-level failures are wrapped in DatabaseError (NATS publish to "<subject>" failed). subscribe(subject: string, handler: NatsMessageHandler): unknown Subscribes to subject using the driver's callback style: it calls connection.subscribe(subject, { callback: handler }). The callback style is chosen over the async-iterator style so a handler can be registered without spawning a consuming loop. Returns the driver's Subscription object — call .unsubscribe() / .drain() on it to stop. Driver failures are wrapped in DatabaseError. request(subject: string, data: NatsPayload, opts?: unknown): Promise<unknown> Sends a request to subject and awaits a single reply. data is encoded per the encoding rules; opts is passed straight through to the driver (e.g. { timeout }). Resolves with the driver's reply message — read .data / .string() / .json() off it. Driver failures are wrapped in DatabaseError. JetStream KV kv(bucket: string): Promise<NatsKvLike> Resolves the JetStream KV bucket view for bucket via connection.jetstream().views.kv(bucket). The resolved view is cached per bucket — subsequent calls for the same bucket return the cached view without re-resolving. The cache is cleared on disconnect(). Failures are wrapped in DatabaseError (NATS KV open bucket "<bucket>" failed). kvPut(bucket: string, key: string, value: NatsPayload): Promise<number> Puts value at key in bucket's KV store (value encoded per the encoding rules). Resolves with the revision number JetStream assigns. Failures are wrapped in DatabaseError. kvGet(bucket: string, key: string): Promise<Uint8Array | null> Gets the raw value stored at key in bucket. Returns the entry's Uint8Array value, or null when the key is absent (or holds a deleted/purged entry). Decode with TextDecoder as needed. Failures are wrapped in DatabaseError. kvDelete(bucket: string, key: string): Promise<void> Deletes key from bucket's KV store. Failures are wrapped in DatabaseError. Example Verification status Unit / mock-verified only. The test suite (tests/nosql/nats.test.ts) is pure unit tests: a mock NatsConnection with publish / subscribe / request spies and a JetStream views.kv() returning a KV mock is injected through the client option. Nothing touches the real nats driver (which is not installed) or the network. Confirmed by the tests: Identity (name/library), isConnected(), getClient(), and idempotent connect() / disconnect(). disconnect() calls the connection's drain(). ConnectionError when using the client / publish() / kvGet() before connecting, and when the lazy require('nats') fails because the driver is absent. publish() UTF-8 encodes strings and passes Uint8Array through unchanged. subscribe() registers the handler via the { callback } option. request() encodes the payload, forwards opts, and returns the reply. kv() resolves via jetstream().views.kv() and caches the view per bucket; operations route to distinct buckets independently. kvPut() encodes and returns the revision; kvGet() returns the entry value or null; kvDelete() routes to the right bucket. Driver errors from publish() and KV operations are wrapped in DatabaseError. No integration testing against a live NATS/JetStream server has been performed. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories