# NATS Lua Implementation - Comprehensive Analysis & TODO

## Version Status

- **Current Lua version**: 1.48.0-lua (from nats.lua)
- **Go reference version**: 1.48.0 (from nats.go)
- **Gap**: 5 versions behind - update needed

## Implementation Status

- **Phase 1 (Critical Bugs)**: ✅ COMPLETED (2025-01-XX)
- **Phase 2 (Architecture Simplification)**: Pending
- **Phase 3 (Feature Parity)**: Pending

---

## FILE DOCUMENTATION - WHY AND HOW TO USE

## 1. nats.lua - Core NATS Client

**WHY**: The main NATS client for basic messaging (connect, publish, subscribe). Handles TCP connection, protocol parsing, message routing.

**HOW TO USE**:

```lua
local nats = require "message/nats/nats"

-- Connect
local nc = nats.connect({host = "localhost", port = 4222})

-- Simple publish
nc:publish("subject", "data")

-- Request-reply
local reply, err = nc:request("subject", "data", 2) -- 2 second timeout

-- Subscribe
nc:subscribe("subject", function(msg)
    print(msg.data)
end)

-- Queue subscribe (load balancing)
nc:queueSubscribe("subject", "queue-group", function(msg)
    print(msg.data)
end)

-- Close
nc:close()
```

**KEY COMPONENTS**:

- `nats.conn` - Connection object (connect, publish, subscribe)
- `nats.msg` - Message object (subject, reply, header, data)
- `nats.header` - Message headers (key-value pairs)
- `nats.subscription` - Subscription object
- `nats.option` - Connection options

---

## 2. nats-api.lua - API Constants & Templates

**WHY**: Central location for all NATS JetStream API subject templates and constants. Prevents typos and ensures consistency.

**HOW TO USE**:

```lua
local api = require "message/nats/nats-api"

-- API subjects are pre-formatted templates
local infoSubject = api.apiAccountInfo  -- "$JS.API.INFO"
local streamCreateSubject = string.format(api.apiStreamCreateT, "mystream")

-- Object Store templates
local objStreamName = string.format(api.objNameTmpl, "mybucket")  -- "OBJ_mybucket"
```

**CONSTANTS DEFINED**:

- JetStream API subjects (stream, consumer, message operations)
- Object Store subject templates (chunks, metadata)
- Default values (chunk size, digest type)

---

## 3. nats-parser.lua - Protocol Parser

**WHY**: Parses NATS wire protocol (MSG, PING, PONG, INFO, ERR, HMSG). High-performance state machine.

**HOW TO USE**: Internal - not called directly. Used by `nats.conn`.

**STATE MACHINE**: Parses incoming bytes line-by-line, tracks state between messages.

---

## 4. nats-nuid.lua - Unique ID Generator

**WHY**: Generate unique identifiers (similar to UUID but faster). Used for inboxes and object names.

**HOW TO USE**:

```lua
local nuid = require "message/nats/nats-nuid"

-- Generate unique ID
local id = nuid.next()  -- Returns 22-character base62 string
```

**ALGORITHM**: Random prefix (12 chars) + sequential counter (10 chars).

---

## 5. nats-error.lua - Error Constants

**WHY**: Centralized error constants for consistency and type safety.

**HOW TO USE**:

```lua
local E = require "message/nats/nats-error"

if err == E.errStreamNotFound then
    -- Handle stream not found
end
```

**ERRORS DEFINED**:

- Stream errors: `errStreamNotFound`, `errStreamNameAlreadyInUse`
- Consumer errors: `errConsumerNotFound`, `errConsumerExists`
- Object Store errors: `errObjectNotFound`, `errDigestMismatch`
- General errors: `errNameRequired`, `errBadObjectMeta`

---

## 6. nats-message-header.lua - Header Decoding

**WHY**: Decode NATS/1.0 protocol headers from messages.

**HOW TO USE**:

```lua
local header = require "message/nats/nats-message-header"

local hdrs, err = header.decodeHeadersMsg(rawHeaderString)
if hdrs then
    print(hdrs["Content-Type"])  -- Access header value
end
```

---

## 7. nats-jetstream.lua - JetStream Context

**WHY**: JetStream API for persistent messaging, streams, consumers, object stores.

**HOW TO USE**:

```lua
local js = nc.js  -- JetStream context attached to connection

-- Publish to JetStream
local ack, err = js:publish("subject", "data")

-- Create Object Store
local objStore, err = js:createObjectStore({bucket = "mybucket"})

-- Get Object Store
local objStore, err = js:objectStore("mybucket")

-- Stream operations (delegated to streamManager)
local stream, err = js:stream("mystream")
local info, err = js:streamInfo("mystream")
```

**KEY FEATURES**:

- `apiRequest(subject, data)` - Low-level API request
- `publishMsg()` - Publish with acknowledgment
- Object Store management
- Stream/Consumer management (via copyMethods from managers)

---

## 8. nats-stream.lua - Stream Operations

**WHY**: High-level stream operations (info, purge, get messages).

**HOW TO USE**:

```lua
local stream = require "message/nats/nats-stream"

local s = stream.new("mystream", js, streamInfo)

-- Get stream info (with pagination support)
local info, err = s:info({subjectFilter = "foo.>", deletedDetails = true})

-- Purge messages
local purged, err = s:purge({subject = "foo.bar", keep = 10})

-- Get message by sequence
local msg, err = s:getMsg(42)

-- Get last message for subject
local msg, err = s:getLastMsgForSubject("foo.bar")

-- Delete message
local ok, err = s:deleteMsg(42)  -- Mark as deleted
local ok, err = s:secureDeleteMsg(42)  -- Overwrite with random data
```

**FEATURES**:

- Cached info (`cachedInfo()`)
- Direct message get support (`allow_direct` in stream config)
- Subject filtering in info requests

---

## 9. nats-consumer.lua - Consumer Operations

**WHY**: Pull and ordered consumer implementations for fetching messages.

**HOW TO USE**:

```lua
local consumer = require "message/nats/nats-consumer"

-- Pull consumer (fetch messages on demand)
local pc = consumer.pullConsumer.new(js, "stream", "consumer-name", nil, info)

-- Fetch single message
local msg, err = pc:next()

-- Fetch batch
local msgs, err = pc:fetch(100)  -- Get up to 100 messages

-- Pull with options
local msgs, err = pc:pull({batch = 100, expires = 5000000000})

-- Messages iterator (auto-fetching)
local iter = pc:messages({pullMaxMessages = 1000})
while true do
    local msg = iter:next()
    if not msg then break end
    -- Process message
end

-- Ordered consumer (automatic recovery)
local oc = consumer.orderedConsumer.new(js, "stream", {
    filterSubjects = {"foo.>"}
})
local iter, err = oc:messages()
```

**PULL CONSUMER**: Fetches messages on demand using `CONSUMER.MSG.NEXT`.

**ORDERED CONSUMER**: Handles consumer recreation on errors, maintains order.

---

## 10. nats-jetstream-manager.lua - Low-Level JetStream API

**WHY**: Direct JetStream API access without abstractions. For advanced users.

**HOW TO USE**:

```lua
local jsm = require "message/nats/nats-jetstream-manager"

local mgr = jsm.new()
mgr.opt = {timeout = 5}
mgr.nc = nc

-- Add stream (low-level)
local resp, err = mgr:addStream({name = "mystream", subjects = {"foo.>"}})

-- Get account info
local info, err = mgr:accountInfo()

-- Purge stream with options
local ok, err = mgr:purgeStream("mystream", {purgeOpts = {subject = "foo.bar"}})

-- Get message (with DirectGet option)
local msg, err = mgr:getMsg("mystream", 42, {directGet = true})

-- Stream name by subject lookup
local streamName, err = mgr:streamNameBySubject("foo.bar")
```

**FEATURES**:

- All CRUD operations for streams and consumers
- Message get/delete
- Account info
- Direct get support
- Stream/consumer listing with pagination

**ISSUES**: See "Simplification Opportunities" section below.

---

## 11. nats-stream-manager.lua - Stream Management

**WHY**: Higher-level stream operations with better error handling.

**HOW TO USE**:

```lua
local streamManager = require "message/nats/nats-stream-manager"

-- Attached to js via copyMethods()
local stream, err = js:createStream({name = "mystream"})

local stream, err = js:updateStream({name = "mystream", max_age = 3600})

local ok, err = js:deleteStream("mystream")

local streams, err = js:listStreams({subject = "foo.>"})

local names, err = js:streamNames({subject = "foo.>"})

-- Get message through stream manager
local msg, err = js:getMsg("mystream", 42)
```

**VALIDATION**: Stream name validation (no spaces, dots, wildcards).

---

## 12. nats-consumer-manager.lua - Consumer Management

**WHY**: Consumer CRUD operations with upsert logic.

**HOW TO USE**:

```lua
local consumerManager = require "message/nats/nats-consumer-manager"

-- Attached to js via copyMethods()

-- Create or update consumer (upsert)
local cons, err = js:createOrUpdateConsumer("stream", {
    name = "myconsumer",
    durable_name = "mydurable",
    ack_policy = "explicit"
})

-- Create only (fails if exists)
local cons, err = js:createConsumer("stream", config)

-- Update only (fails if not exists)
local cons, err = js:updateConsumer("stream", config)

-- Delete
local ok, err = js:deleteConsumer("stream", "consumer")

-- List consumers (with pagination)
local consumers, err = js:listConsumers("stream")
local names, err = js:consumerNames("stream")

-- Pause/unpause consumer
local ok, err = js:pauseConsumer("stream", "consumer", pauseUntil)
local ok, err = js:resumeConsumer("stream", "consumer")

-- Unpin consumer
local ok, err = js:unpinConsumer("stream", "consumer", "group")
```

**FEATURES**:

- `copyMethods()` pattern to copy functions to jetstream object
- Automatic upsert (create if not exists, update if config matches)
- Pagination support for list operations

---

## 13. nats-object-store.lua - Object Store Implementation

**WHY**: Blob store for large files using JetStream streams.

**HOW TO USE**:

```lua
local objStore = require "message/nats/nats-object-store"

-- Attached to stream via copyMethods()

-- Put object
local info, err = objStore:put({
    name = "myfile.txt",
    description = "My file",
    metadata = {key = "value"}
}, "file data here")

-- Put bytes/string
info, err = objStore:putBytes("filename", data)
info, err = objStore:putString("filename", "string data")

-- Get object
local data, info, err = objStore:get("filename")
local bytes, err = objStore:getBytes("filename")
local str, err = objStore:getString("filename")

-- Get info only
local info, err = objStore:getInfo("filename", {showDeleted = true})

-- Get with ordered consumer (better for large objects)
local data, info, err = objStore:getOrderedConsumer("filename")

-- Update metadata
local ok, err = objStore:updateMeta("oldname", {name = "newname"})

-- Delete
local ok, err = objStore:delete("filename")

-- Add link (to another object or bucket)
local info, err = objStore:addLink("linkname", otherObjectInfo)
local info, err = objStore:addBucketLink("bucketlink", otherBucket)

-- List objects
local objs, err = objStore:list({showDeleted = false})

-- Seal bucket (no more modifications)
local ok, err = objStore:seal()

-- Status
local status, err = objStore:status()
```

**CHUNKING**: Large files split into 128KB chunks (configurable).

**DIGEST**: SHA-256 hash for data integrity (currently DISABLED - see bugs).

---

## LINE-BY-LINE COMPARISON: LUA vs GO

## Missing Features from Go 1.48.0

### 1. Async Publish with Ack Futures (js.go:47-72)

**Go has**:

```go
PublishAsync(subj string, data []byte, opts ...PubOpt) (PubAckFuture, error)
PublishAsyncPending() int
PublishAsyncComplete() <-chan struct{}
```

**Lua**: Missing entirely. No async publish tracking.

**Impact**: Cannot track pending publishes in production systems.

**Recommendation**: Implement using coroutines instead of Go channels.

---

### 2. Context/Timeout Support (Throughout Go code)

**Go has**:

```go
func (js *js) apiRequestWithContext(ctx context.Context, subj string, req []byte) (*Msg, error)
```

**Lua**: Uses simple timeout values. No cancellation.

**Impact**: Cannot cancel long-running requests.

**Recommendation**: Lua coroutines with cooperative multitasking don't need context cancellation. Current timeout approach is fine.

---

### 3. Direct Get API Options (jsm.go:56-63, 413-421)

**Go has**:

```go
func DirectGet() JSOpt
func DirectGetNext(subject string) JSOpt
```

**Lua**: Partially implemented in `_getMsg()` method.

**Missing**: `DirectGetNext` option for wildcard subject searching.

---

### 4. Stream Purge Options (jsm.go:387-390)

**Go has**:

```go
func (s *StreamPurgeRequest) configureJSContext(js *jsOpts) error
```

**Lua**: Has `purgeOpts` but not fully integrated into all API paths.

---

### 5. Stream Info Options (jsm.go:392-395)

**Go has**:

```go
func (s *StreamInfoRequest) configureJSContext(js *jsOpts) error
```

**Lua**: Partially implemented in `stream.lua:stream()`.

---

### 6. Object Store Watch (object.go:90)

**Go has**:

```go
Watch(opts ...WatchOpt) (ObjectWatcher, error)
```

**Lua**: Missing entirely.

**Impact**: Cannot monitor object store for changes.

---

### 7. Subject Transform (jsm.go:256-260)

**Go has**:

```go
SubjectTransform *SubjectTransformConfig `json:"subject_transform,omitempty"`
```

**Lua**: Not implemented.

---

### 8. RePublish (jsm.go:262-268)

**Go has**:

```go
RePublish *RePublish `json:"republish,omitempty"`
```

**Lua**: Not implemented.

---

### 9. Compression Support (partial)

**Go has**:

```go
Compression StoreCompression `json:"compression"`
```

**Lua**: Has `compression = "s2"` string but no enum constants.

---

### 10. CheckStreamName Validation (jsm.go:573-581)

**Go**:

```go
func invalidStreamName(stream string) error {
    if stream == _EMPTY_ {
        return ErrStreamNameRequired
    }
    if strings.ContainsAny(stream, ". ") {
        return ErrInvalidStreamName
    }
    return nil
}
```

**Lua** (nats-stream-manager.lua:26-33):

```lua
local function validateStreamName(name)
    if not name or name == "" then
        return nil, E.errStreamNameRequired
    end
    if string.find(name, "[>*. /\\']") then
        return nil, E.errInvalidStreamName
    end
end
```

**Difference**: Lua checks for more characters (`>*. /\\'` vs Go's just `.`). This is MORE CORRECT for stream names.

---

## CODE ISSUES FOUND

## Critical Bugs

### 1. Hash Verification Disabled (nats-object-store.lua:210) ✅ FIXED

```lua
local checkHash = false -- nada, hash calculation does not work correctly
```

**Impact**: Data integrity is NOT verified when retrieving objects. Corrupted data will be accepted.

**Go Reference**: object.go has full hash verification.

**Fix Applied**: Changed `checkHash = true` on line 202 to enable SHA-256 digest verification.

**Status**: ✅ Completed - Hash verification now enabled by default.

---

### 2. Module-Level Variable Anti-Pattern (nats-consumer.lua:29) ✅ FIXED

```lua
local _, err  -- Declared at module level
```

**Impact**: If two coroutines call these functions concurrently, `err` gets overwritten.

**Fix**: Each function should declare its own `err` variable.

**Status**: ✅ Already fixed in earlier edits.

---

### 3. Global Variables in Parser (nats-parser.lua:30) ✅ FIXED

```lua
local ps, ma -- temp variables to prevent gc
local arg, err, i, bufLen
local b  -- for splitArgs
```

**Impact**: Thread-safety issues in coroutine environment.

**Fix Applied**: Made all variables local to their respective functions:

- `ps`, `ma` → local to `cloneMsgArg()`, `processMsgArgs()`, `processHeaderMsgArgs()`, `parse()`
- `arg`, `err`, `i`, `bufLen`, `b` → local to `parse()`
- `b` → local to `splitArgs()`

**Status**: ✅ Completed - All module-level variables removed.

---

## Architecture Issues

### 1. copyMethods Pattern (All -manager files)

**Current Pattern**:

```lua
function streamManager.copyMethods(tbl, copyStream)
    for functionName, func in pairs(streamManager) do
        if tbl[functionName] == nil and functionName ~= "copyMethods" then
            tbl[functionName] = func
        end
    end
end

-- Usage:
streamManager.copyMethods(jetstream)
streamManager.copyMethods(objectStore, true)
consumerManager.copyMethods(jetstream)
```

**Problems**:

- Creates hidden dependencies
- Hard to see what methods an object has without running code
- Name collision risk
- Not explicit (violates "explicit is better than implicit")

**Simplification**:

```lua
-- Option 1: Direct composition
jetstream.addStream = streamManager.addStream
jetstream.updateStream = streamManager.updateStream

-- Option 2: Delegation
function jetstream:addStream(cfg)
    return streamManager.addStream(self, cfg)
end

-- Option 3: Mixin table
local methods = {}
for k, v in pairs(streamManager) do
    if type(v) == "function" then
        methods[k] = v
    end
end
setmetatable(jetstream, {__index = methods})
```

---

### 2. Redundant Error Checking

**Example** (nats-object-store.lua:64-70, now fixed):

```lua
m, err = self.stream:getLastMsgForSubject(metaSubj)
if err then
    if err == E.errObjectNotFound then  -- REDUNDANT
        return nil, E.errObjectNotFound
    end
    return nil, err
end
```

**Simpler**:

```lua
m, err = self.stream:getLastMsgForSubject(metaSubj)
if err then
    return nil, err  -- Let caller handle
end
```

---

### 4. table.insert Usage

**Bad** (older code):

```lua
table.insert(arr, obj)
```

**Good** (Lua best practice):

```lua
arr[#arr + 1] = obj
```

**Status**: Already fixed in earlier edits.

---

## SIMPLIFICATION OPPORTUNITIES

## 1. Remove copyMethods Pattern

**Current State**:

- `nats-jetstream-manager.lua`: 600+ lines
- `nats-stream-manager.lua`: 150+ lines
- `nats-consumer-manager.lua`: 250+ lines

**Simplification**:

**Step 1**: Create shared API module

```lua
-- nats-api-common.lua (NEW FILE)
local apiCommon = {}

function apiCommon.request(js, subject, data, opts)
    -- Consolidated API request logic
    local timeout = (opts and opts.timeout) or js.opts.timeout or 5
    local reqJson
    if data then
        local err
        reqJson, err = toJsonRaw(data)
        if err then
            return nil, err
        end
    else
        reqJson = "{}"
    end

    local resp, err = js.nc:request(subject, reqJson, timeout)
    if err then
        return nil, err
    end

    local result, jsonErr = fromJson(resp.data)
    if jsonErr then
        return nil, jsonErr
    end

    if result.error then
        return nil, result.error.description or result.error
    end

    return result
end

return apiCommon
```

**Step 2**: Simplify jetstream manager

```lua
-- Instead of copyMethods, use direct delegation
local streamManager = require "message/nats/nats-stream-manager"

function jetstream:addStream(cfg)
    return streamManager.addStream(self, cfg)
end

function jetstream:updateStream(cfg)
    return streamManager.updateStream(self, cfg)
end

-- Or even simpler - just alias the module
jetstream.streamManager = streamManager
```

**Savings**: ~200 lines of complex code

---

## 2. Consolidate Validation Logic

**Current**: Scattered across files

- `nats-stream-manager.lua:validateStreamName()`
- `nats-consumer-manager.lua` (inline checks)
- `nats-jetstream-manager.lua:invalidStreamName()` and `invalidConsumerName()`

**Simplification**:

```lua
-- nats-validation.lua (NEW FILE)
local validation = {}

function validation.streamName(name)
    if not name or name == "" then
        return nil, E.errStreamNameRequired
    end
    if string.find(name, "[>*. /\\' ]") then
        return nil, E.errInvalidStreamName
    end
end

function validation.consumerName(name)
    if not name or name == "" then
        return nil, E.errConsumerNameRequired
    end
    if string.find(name, "[.* ]") then
        return nil, E.errInvalidConsumerName
    end
end

function validation.bucketName(name)
    if not name or name == "" then
        return nil, E.errBucketNameRequired
    end
    if string.find(name, "[>*. /\\']") then
        return nil, E.errInvalidBucketName
    end
end

return validation
```

---

## 3. Simplify List Operations Pagination

**Current**: Every list function has same pagination loop:

```lua
local offset = 0
while true do
    req.offset = offset
    resp, err = self:apiRequest(subject, req)
    -- ...
    if resp.total == 0 or (resp.offset + resp.limit) >= resp.total then
        break
    end
    offset = resp.offset + resp.limit
end
```

**Simplification**:

```lua
-- nats-api-common.lua
function apiCommon.paginatedRequest(js, subject, req, opts)
    local results = {}
    local offset = 0

    while true do
        local pageReq = opts.transformRequest and opts.transformRequest(req, offset) or req
        pageReq.offset = offset

        local resp, err = apiCommon.request(js, subject, pageReq)
        if err then
            return nil, err
        end

        local items = resp[opts.resultField] or resp
        if items then
            for i = 1, #items do
                results[#results + 1] = items[i]
            end
        end

        if resp.total == 0 or (resp.offset + resp.limit) >= resp.total then
            break
        end
        offset = resp.offset + resp.limit
    end

    return results
end

-- Usage:
function streamManager:listStreams(listOpt)
    local req = {offset = listOpt and listOpt.offset or 0}
    if listOpt and listOpt.subject then
        req.subject = listOpt.subject
    end
    return apiCommon.paginatedRequest(self, api.apiStreamListT, req, {
        resultField = "streams"
    })
end
```

---

## 4. Remove Module-Level Variables

**Files to fix**:

- `nats-consumer.lua:29` - `local _, err` (already fixed)
- `nats-parser.lua:30` - `local ps, ma`

**Fix for parser**:

```lua
function parser:parse(buf)
    local ps = self.ps  -- Local to function
    local ma = ps.ma
    -- ... rest of code
end
```

---

## MISSING Go FEATURES (By Priority)

## High Priority (Production Impact)

1. **Async Publish Ack Tracking**
   - Go: `PublishAsync()` with `PubAckFuture`
   - Lua: Not implemented
   - Impact: Cannot track publish confirmations in high-throughput scenarios
   - Recommendation: Implement using Lua coroutines

2. **Object Store Watch**
   - Go: `Watch(opts ...WatchOpt) (ObjectWatcher, error)`
   - Lua: Not implemented
   - Impact: Cannot monitor object store for changes
   - Recommendation: Implement with Lua iterator pattern

3. **Hash Verification**
   - Go: Full SHA-256 digest verification
   - Lua: Disabled (`checkHash = false`)
   - Impact: Data integrity at risk
   - Recommendation: Fix the digest calculation

## Medium Priority (Feature Parity)

1. **DirectGetNext Option**
   - Go: `DirectGetNext(subject string) JSOpt`
   - Lua: Not implemented
   - Impact: Cannot efficiently search for next message by subject

2. **Subject Transform**
   - Go: `SubjectTransformConfig`
   - Lua: Not implemented
   - Impact: Cannot transform subjects before storage

3. **RePublish**
   - Go: `RePublish` configuration
   - Lua: Not implemented
   - Impact: Cannot automatically republish messages

## Low Priority (Nice to Have)

1. **Compression Constants**
   - Go: `StoreCompression` enum
   - Lua: Has `"s2"` string but no constants

2. **Stream List Filter Option**
   - Go: `StreamListFilter(subject string)`
   - Lua: Partially implemented

---

## LUA COROUTINE ADVANTAGE

Since Lua uses coroutines (cooperative multitasking, single-threaded), much Go complexity can be avoided:

## NOT NEEDED in Lua

1. **Mutexes** - No concurrent access to shared state
2. **Channels** - Use direct function calls and coroutines
3. **WaitGroups** - Sequential execution is natural
4. **Context cancellation** - Cooperative scheduling simplifies this

## BETTER in Lua

1. **Error handling** - No need for special atomic operations
2. **State management** - No race conditions
3. **Memory management** - Predictable GC

**This means**: The Lua code CAN and SHOULD be much simpler than the Go equivalent.

---

## IMPLEMENTATION PLAN

## Phase 1: Critical Bugs ✅ COMPLETED

1. ✅ Fix hash verification in `nats-object-store.lua` - Changed `checkHash = true`
2. ✅ Remove module-level variables - Fixed in `nats-parser.lua` and `nats-consumer.lua`
3. ✅ Fix all remaining error handling inconsistencies - Fixed JSON encode error in `nats-consumer.lua`, improved purge error handling in `nats-stream.lua`, added `opts.subject` support to `getMsg()`

**Files Modified**:

- `/Volumes/nc/nc-server/lib/message/nats/nats-object-store.lua` - Enabled hash verification
- `/Volumes/nc/nc-server/lib/message/nats/nats-parser.lua` - Removed all module-level variables
- `/Volumes/nc/nc-server/lib/message/nats/nats-consumer.lua` - Fixed JSON error handling
- `/Volumes/nc/nc-server/lib/message/nats/nats-stream.lua` - Improved error handling, added `opts.subject` support

## Phase 2: Architecture Simplification

1. Create `nats-api-common.lua` for shared API logic
2. Create `nats-validation.lua` for validation functions
3. Remove `copyMethods()` pattern
4. Use direct method delegation instead

## Phase 3: Feature Parity

1. Implement async publish tracking
2. Implement ObjectStore:watch()
3. Add DirectGetNext option
4. Add Subject Transform support

---

## TESTING NOTES

## Current Test Files

1. **nats-test.lua**: Basic JetStream operations
   - Creates object store
   - Tests bucket:get() with metadata and content
   - Lists streams and consumers

2. **nats-tcp-test.lua**: Performance testing
   - Tests large datasets (loopCount parameter)
   - Measures messages/second throughput
   - Tests object put/get/delete operations

**Missing Tests**:

- Hash verification
- Error handling paths
- Manager method simplification
- Pagination edge cases
- Ordered consumer recovery scenarios

---

---

## CONSOLIDATION STATUS (2025-12-27)

## Completed: nats-jsm.lua → nats-jetstream.lua

All consumer and stream management functions from `nats-jsm.lua` have been consolidated into `nats-jetstream.lua`:

**Added to nats-jetstream.lua:**

- Helpers: `invalidStreamName()`, `invalidConsumerName()`, `getJsOpts()`, `listAll()`
- Stream CRUD (with validation + `opt` parameter): `createStream()`, `updateStream()`, `deleteStream()`, `streamInfo()`, `createOrUpdateStream()`
- Consumer CRUD: `_upsertConsumer()`, `addConsumer()`, `updateConsumer()`, `deleteConsumer()`, `consumerInfo()`, `consumers()`, `consumerNames()`
- Message ops: `purgeStream()`, `getMsg()`, `getLastMsg()`, `deleteMsg()`, `secureDeleteMsg()`, `_deleteMsg()`
- Advanced consumer: `createOrUpdateConsumer()`, `orderedConsumer()`, `pauseConsumer()`, `resumeConsumer()`
- Utilities: `accountInfo()`, `streamNameBySubject()`
- Requires added: `nats-pull-consumer`, `nats-ordered-consumer`

**Status**:

- `nats-jsm.lua` - Can now be deleted (all functionality moved to nats-jetstream.lua)
- `nats-stream.lua` - Can now be deleted (methods duplicated in nats-jetstream.lua)

---

## MISSING from Go's NEW jetstream/ package

The Lua implementation copied from `nats-jsm.lua` (Go's OLD/LEGACY jsm.go), but Go's NEW `jetstream/` package has additional features:

### 1. Key-Value Store (kv.go - 1556 lines)

**Go has:**

```go
// Create or get KV store
kv, err := js.CreateOrUpdateKeyValue(ctx, "mykv")
kv, err := js.KeyValue(ctx, "mykv")

// KV operations
kv.Put(ctx, "key", "value")
kv.Get(ctx, "key")
kv.Delete(ctx, "key")
kv.Purge(ctx, opts...)
kv.Status(ctx)

// Watch for changes
watcher, err := kv.Watch(ctx, "key", ...)
```

**Lua**: Completely missing. No `nats-kv.lua` file exists.

**Impact**: KV is a major feature - essentially a distributed database built on JetStream.

---

### 2. Push Consumer (push.go - 268 lines)

**Go has:**

```go
// Push consumer delivers messages to a subscription
push, err := js.CreateOrUpdatePushConsumer(ctx, stream, cfg)
push, err := js.PushConsumer(ctx, stream, name)

// Push consumer info
info, err := push.Info()
```

**Lua**: Missing. Only pull consumers are implemented (`nats-pull-consumer.lua`).

**Difference**: Pull consumers fetch on demand; Push consumers deliver to a subscription.

---

### 3. Consumer Interface (consumer.go - 541 lines)

**Go has Consumer interface with:**

```go
type Consumer interface {
    ConsumerConfig() (ConsumerConfig, error)
    Info() (ConsumerInfo, error)
    CachedInfo() ConsumerInfo

    // Message consumption
    Fetch(n int, opts ...FetchOpt) (MsgSeqPair, error)
    FetchBytes(batch int, opts ...FetchOpt) ([]*Msg, error)
    Consume(opts ...ConsumeOpt) (MsgConsumer, error)
    Messages(opts ...MessagesOpt) (MessageConsumer, error)

    // Next message
    Next(msg Msg) error

    // Consumer operations
    IsPullMode() bool
    Close() error
}
```

**Lua**: Consumer objects are returned but don't have full interface implementation.

---

### 4. Stream.UnpinConsumer()

**Go has (stream.go):**

```go
func (s *stream) UnpinConsumer(ctx context.Context, consumer string, group string) error
```

**Lua**: `nats-jsm.lua` has `unpinConsumer()` but NOT copied to `nats-jetstream.lua`.

---

## Recommendation

1. **Create `nats-kv.lua`** - Implement Key-Value store (high priority, 1500+ lines in Go)
2. **Add Push Consumer support** - Create `nats-push-consumer.lua`
3. **Copy `unpinConsumer()`** to `nats-jetstream.lua` if needed
4. **Enhance Consumer interface** - Add missing methods to consumer objects

---

## MISSING FEATURES FROM Go jetstream/ (2025-12-29)

Comparison between:

- `/Volumes/nc/nc-server/lib/message/nats/nats-ordered-consumer.lua`
- `/Volumes/nc/nc-server/lib/message/nats/nats-pull-consumer.lua`
- `/Volumes/stu/Documents/installed/go/nats.go/jetstream/ordered.go`
- `/Volumes/stu/Documents/installed/go/nats.go/jetstream/pull.go`

### ordered-consumer.lua vs Go ordered.go

Missing in Lua:

1. `fetchBytes(maxBytes, opts)` - Fetch by byte size limit (Go:447-485)
2. `fetchNoWait(batch)` - Non-blocking fetch variant (Go:487-513)
3. `maxResetAttempts` config - Retry with backoff (Go:620-622)
4. `retryWithBackoff()` - Exponential backoff for consumer creation (Go:769-828)
5. `stopAfter` functionality - Auto-stop after N messages (Go:104-111, 304-313)
6. Reconnect notification handlers (Go:687-699)
7. `info()` and `cachedInfo()` methods (Go:701-740) - Lua has `info()` but not full implementation
8. `consume()` method - Callback-based consumption (Go:82-201)
9. `closed()` method on orderedSubscription (Go:385-405)

### pull-consumer.lua vs Go pull.go

Missing in Lua:

1. `fetchBytes(maxBytes, opts)` (Go:854-886)
2. `fetchNoWait(batch)` (Go:888-898)
3. `stopAfter` functionality (Go:274-277, 475-477, 739-740, 774-780)
4. Priority groups (Go:218-228, 512-522)
5. `minPending` and `minAckPending` options (Go:103-104, 333, 404, 409)
6. PinID support (Go:105, 290-292, 467-482, 673-674)
7. Context support for `next()`/`fetch()` (Go:597-619, 828-836, 863-871)
8. `closed()` method (Go:805-818)
9. Advanced heartbeat monitoring (Go:568-576, 631-636, 1068-1077)
10. Connection status monitoring (Go:243, 343-388, 539, 564-581)
11. `drain()` method (Go:787-800)
12. Reconnect handlers (Go:364-383)

### Key Architectural Differences

| Feature | Go | Lua |
|---------|-----|-----|
| Concurrency | Goroutines + channels | None (single-threaded) |
| Pull loop | `go pullMessages()` (Go:1041) | `while not do` in `pull()` (Lua:109-125) |
| Message buffering | `msgs chan *nats.Msg` (Go:527) | `receivedMsgs` array (Lua:49) |
| Heartbeat timer | `time.AfterFunc` (Go:1073) | Timeout loop (Lua:109-125) |
| Pending messages | `pendingMsgs` struct with thresholds (Go:154-157, 456-499) | Simple count check (Lua:67-68) |

### High Priority Missing Features

1. **`drain()` method** - Graceful shutdown, process buffered messages
2. **`fetchBytes()` method** - Fetch by byte size instead of count
3. **`fetchNoWait()` method** - Non-blocking fetch
4. **`stopAfter` functionality** - Auto-stop after N messages
5. **PinID support** - Consumer pinning for message ordering

---

## END OF DOCUMENTATION

Last updated: 2025-12-29
Status: Phase 1 (Critical Bugs) completed - Phase 2 (Architecture Simplification) - nats-jsm.lua consolidated into nats-jetstream.lua - Added drain() method to consumers
