Skip to content

NATS

NATS is used as the internal pub/sub message bus. Today it's used only within CoreAPI itself (published and subscribed by the same process), but the client wrapper and infrastructure are already positioned for cross-service use later — e.g. other apps subscribing to media/channel/schedule events without going through CoreAPI's HTTP API.

Current status: Active, but narrow. Events are published and logged; nothing yet consumes them to drive real behavior (no websocket push to frontends, no cross-service side effects). See Current Usage below for specifics.

Compose Service

services:
  nats:
    image: nats:alpine
    container_name: nats
    ports:
      - "4222:4222"   # client connections
      - "8222:8222"   # HTTP monitoring
    networks:
      - internal
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "wget", "-qO-", "http://localhost:8222/healthz"]
      interval: 5s
      timeout: 3s
      retries: 5
Port Purpose
4222 Client connections — this is what natsclient.Init() connects to
8222 HTTP monitoring endpoint — used by the healthcheck (/healthz), also browsable directly for connection/subscription stats

Environment Variables

Variable Value Role Status
NATS_URL nats://nats:4222 Client connection string. Falls back to this same value in code if unset — see natsclient.Init(). Active

That's the only NATS-specific env var in the system today. No auth, no TLS, no clustering config — this is a minimal single-node setup appropriate for local/dev. Production hardening (auth tokens, TLS, clustering) isn't configured yet and would need to be added before any external-facing or multi-tenant use.

Client Wrapper

CoreAPI wraps nats.go in a small package: pkg/natsclient.

package natsclient

var NC *nats.Conn

func Init()                                          // connects, called once in main.go
func Publish(subject string, data interface{}) error  // JSON-marshals data, publishes
func Subscribe(subject string, handler func([]byte)) (*nats.Subscription, error)
func Close()                                          // called via defer in main.go

Wiring in main.go

natsclient.Init()
defer natsclient.Close()

Initialized once at startup, alongside Postgres, Redis, SeaweedFS, and AuthAPI — treated as a core infra dependency, not optional.

How to publish

import "mashup.castis.io/playtelly/CoreAPI/pkg/natsclient"

natsclient.Publish("my.event.name", map[string]interface{}{
    "id":     123,
    "status": "done",
})

Publish JSON-marshals whatever you pass as data — no schema enforcement, so subject naming and payload shape are a convention, not a contract. Keep payloads consistent per-subject by hand.

How to subscribe

natsclient.Subscribe("my.event.name", func(data []byte) {
    var e MyEventPayload
    json.Unmarshal(data, &e)
    // handle it
})

Typically registered once at startup via a RegisterSubscribers() function per domain (see media domain's handler_nats.go for the existing pattern).

Current Usage (CoreAPI)

The media domain is the only current publisher/subscriber. Three events are published during the upload-processing lifecycle:

Subject Published when Subscribed — currently does
media.processing.started POST /media/:id/process called, or tusd webhook fires Logs only
media.ready Filer move to final path completes, status → ready Logs only
media.thumbnail.ready Thumbnail generated and uploaded to SeaweedFS Logs only

Both the publish and subscribe sides live in the same CoreAPI process today — there's no cross-service consumption yet. The subscriber handlers (domain/media/handlers/handler_nats.go) have explicit // future: comments marking intended next steps:

// future: notify frontend via WebSocket bridge
// future: provisioning.UnlockEntitlement(e.TenantID, e.MediaID)
// future: analytics.RecordUpload(e.TenantID, e.MediaID)

Practical implication: MediaList.jsx's live-update behavior (detecting when a processing item becomes ready) is driven by polling (setInterval, every 3–4s), not by NATS. NATS events fire correctly and are logged server-side, but nothing today pushes them to the frontend. If real-time push ever replaces polling, this is the mechanism it would build on top of — the publish side already exists.

Verifying NATS is running

# health check
curl http://localhost:8222/healthz

# connection/subscription stats
curl http://localhost:8222/connz
curl http://localhost:8222/subsz

subsz is useful for confirming a subscriber actually registered — if handler_nats.go's RegisterSubscribers() didn't get called for some reason, you'd see zero subscriptions for the media.* subjects here even though publishing wouldn't error (NATS doesn't require a subscriber to exist for Publish to succeed — messages with no subscribers are simply dropped, no error, no queue, no retry).

Design notes for future cross-service use

A few things worth deciding before other apps start subscribing, since none of this is enforced today:

  • Subject naming convention — currently media.<event>, dot-delimited, domain-prefixed. Worth keeping consistent (distribution.<event>, playback.<event>, etc.) if other domains start publishing.
  • No delivery guarantees — this is core NATS (fire-and-forget), not JetStream. If a subscriber is down when an event publishes, that event is lost — no replay, no persistence. Fine for today's "log it" use case; would need JetStream (or a different subject/durable-consumer setup) if a future consumer needs guaranteed delivery (e.g. billing, entitlement unlocking).
  • No schema/versioning — payloads are ad-hoc map[string]interface{} or anonymous structs per publish call. If cross-service consumers appear, worth formalizing payload shapes (shared Go structs, or a schema doc) so a publisher-side field rename doesn't silently break a consumer in a different repo.