Austin's SWE Notes 🧠

NATS

To define a stream & consumer

import { connect, nanos, AckPolicy } from "nats";


const nc = await connect({ servers: "localhost:4222" })
const jsm = await nc.jetstreamManager();


await jsm.streams.add({ name: "DEPOSITS", subjects: ["deposits.>"] });

await jsm.consumers.add("DEPOSITS", {
  durable_name: "worker",
  ack_policy: AckPolicy.Explicit,
  ack_wait: nanos(10_000),
  max_deliver: 5
})

console.log("Done");
await nc.drain();

nc.drain() here is a graceful shutdown, stop taking new message and stuff.

AckPolicy.Explicit = ack each message individually.
AckPolicy.All = once ack a message, previous messages also acked.

deposits.> means everything after this, says subject could be deposits.something.something_else
To have only 1 thing, we can have deposits.* which match deposits.something but not deposits.something.something_else

This also means our consumer will consume all deposits related event. If we want a specific subject, we can do like so

await jsm.consumers.add("DEPOSITS", {
  durable_name: "bonus-worker",
  filter_subject: "deposits.created",   // only this subject
  ack_policy: AckPolicy.Explicit,
  ack_wait: nanos(10_000),
  max_deliver: 5,
});

To create a worker and consume the stream

import { connect, JSONCodec } from "nats";

import { applyDepositEvent, type DepositEvent } from "./db.js";

const nc = await connect({ servers: "localhost:4222" });
const js = nc.jetstream();
const jc = JSONCodec<DepositEvent>();

const consumer = await js.consumers.get("DEPOSITS", "worker");
const messages = await consumer.consume();

console.log("worker waiting for message...");

for await (const m of messages) {
  if (m.subject !== "deposits.created") { m.term(); continue; }
  
  try {
    const event = jc.decode(m.data);
    if (process.env.FAIL) throw new Error("boom");

    const result = await applyDepositEvent(event);

    console.log(event.eventId, result, `delivery #${m.info.deliveryCount}`);
    m.ack();
  } catch (err) {
    console.error("failed with retry in 5s", err);
    m.nak(5000);
  }
}

We use JSONCodec here since NATs only understand byte, JSONCodec helps to encode JSON → Byte → JSON

m.term() means never send this message to me again

To create a producer produce to stream

import { connect, JSONCodec } from "nats";

const nc = await connect({ servers: "localhost:4222" });
const js = nc.jetstream();
const jc = JSONCodec();

const eventId = process.argv[2] ?? crypto.randomUUID();

const ack = await js.publish(
  "deposits.created",
  jc.encode({ eventId, playerId: "p1", amountCents: 5000 }),
  { msgID: eventId }
)

console.log({ eventId, seq: ack.seq, duplicate: ack.duplicate })

await nc.drain()

Manage and view the stream

To manage and view the stream, use nats cli

← Decorator
Prisma →