# NATS Consumer Patterns for Object Store

## Overview

This document explains how NATS consumers work for object store operations, comparing the Go reference implementation with the Lua implementation.

## How Go's ObjectStore.Get() Works

### Step-by-Step Process

```go
// From /Volumes/stu/Documents/installed/go/nats.go/jetstream/object.go:827
func (obs *obs) Get(ctx context.Context, name string, opts ...GetObjectOpt) (ObjectResult, error) {
    // Step 1: Get metadata (contains nuid)
    info, err := obs.GetInfo(ctx, name, infoOpts...)

    // Step 2: Create chunk subject from nuid
    chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, info.NUID)
    // Example: $O.bucket.C.sfagbUj7acRl3ZYrGD58nO

    // Step 3: Subscribe with OrderedConsumer
    subscribeOpts := []nats.SubOpt{
        nats.OrderedConsumer(),
        nats.BindStream(streamName),
    }
    obs.pushJS.Subscribe(chunkSubj, processChunk, subscribeOpts...)

    // Step 4: Stream chunks via callback as they arrive
    // processChunk writes to net.Pipe(), calculates SHA256
}
```

### Key Characteristics

| Aspect | Go Implementation | Notes |
|--------|------------------|-------|
| Gets full data? | YES | Streams all chunks to io.Reader |
| Gets metadata? | YES | First call to GetInfo() |
| Uses wildcard? | NO | Filters by specific chunk subject with nuid |
| Creates consumers? | YES | Ephemeral consumers (ord_1, ord_2...) |
| Deletes consumers? | NO | Never deletes - auto-cleanup by server |
| Consumer type | OrderedConsumer | Subscription-level option |

## How OrderedConsumer Works in Go

### It's a Subscription Option, Not a Consumer Type

```go
// From js.go:2495
func OrderedConsumer() SubOpt {
    return subOptFn(func(opts *subOpts) error {
        opts.ordered = true
        return nil
    })
}
```

When you subscribe with `OrderedConsumer()`:

1. Creates ephemeral consumer with random name (ord_1, ord_2, etc.)
2. Tracks cursor position (streamSeq, deliverSeq) internally
3. Auto-resets on sequence mismatch using `by_start_sequence`
4. Never deletes old consumers (server auto-cleanup)

### Consumer Creation Pattern

```go
// From js.go:2203
func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
    // Creates NEW consumer with:
    // - name: ord_<serial> (incrementing)
    // - opt_start_seq: sseq (uses by_start_sequence internally)
    // - Rewires subscription to new consumer
    // - Old consumer left for server to clean up
}
```

**Key point**: Old consumers are NEVER deleted. They're ephemeral (inactive_threshold = 5 min) and auto-cleaned by NATS server.

## Lua Implementation Comparison

### Current getOrderedConsumer()

```lua
-- Current approach
local chunkSubject = format(api.objChunksPreTmpl, self.name, info.nuid)
local oc = self.orderedConsumerCache[chunkSubject]

if oc == nil then
    oc = self.js:orderedConsumer(self.streamName, {filter_subject = chunkSubject})
    self.orderedConsumerCache[chunkSubject] = oc
end

local msgs = oc:fetch(chunkCount, opt)
-- Process chunks...
```

### Current getDirect()

```lua
-- From nats-object-store.lua:319-390
local chunkSubject = format(api.objChunksPreTmpl, self.name, info.nuid)
local cons = self.directConsumerCache[chunkSubject]

if cons == nil then
    local newNuid = nuidNext()
    local consumerName = "object-store-direct-" .. newNuid:sub(1, 8)
    local consumerCfg = {
        name = consumerName,
        filter_subject = chunkSubject,
        ack_policy = "none",
        replay_policy = "instant"
    }
    cons = self.js:createOrUpdateConsumer(self.streamName, consumerCfg)
    self.directConsumerCache[chunkSubject] = cons  -- CACHED!
end

-- Fetch chunks... NO deleteConsumer call!
```

**ACTUAL behavior**: Creates consumer with random name, CACHES it by chunkSubject, NEVER deletes (only on bucket purge).

For 50 unique objects = 50 consumer operations (create only), cached for reuse.

## Should We Use Wildcard for Single-Object get()?

### NO - For Single Object, Use Specific Subject

**Wildcard approach** (what NOT to do):

```lua
-- This would be inefficient - NOW BLOCKED BY CODE!
local oc = self.js:orderedConsumer(self.streamName, {
    filter_subject = "$O.bucket.C.>",  -- ALL chunks from ALL objects
    deliverPolicy = "all"  -- ERROR: wildcard + "all" is not allowed!
})
-- Returns error: "deliverPolicy 'all' with wildcard filter_subject is not allowed"

-- Correct: omit deliverPolicy to use default "by_start_sequence"
local oc = self.js:orderedConsumer(self.streamName, {
    filter_subject = "$O.bucket.C.>"  -- Uses default by_start_sequence
})
```

**Why wildcard + "all" is bad**:

1. `deliver_policy = "all"` reads from sequence 1 every time
2. Fetches ALL data from ALL objects repeatedly
3. Need to sort/group by nuid after fetching
4. Memory intensive
5. Stream keeps growing - never reaches end
6. Defeats the purpose of subject-based filtering

**What about wildcard + "new"?** Also bad:

- `deliver_policy = "new"` only gets messages AFTER consumer creation
- Misses all existing data in the stream

**What about wildcard + "by_start_sequence"?** Complex:

- Requires external cursor tracking
- Need to know exact sequence positions
- Complex state management

**Correct approach** (what Go does):

```lua
-- Filter by SPECIFIC chunk subject for this object
local chunkSubject = "$O.bucket.C.sfagbUj7acRl3ZYrGD58nO"  -- Specific nuid!
local oc = self.js:orderedConsumer(self.streamName, {
    filter_subject = chunkSubject  -- No deliverPolicy needed, uses default "by_start_sequence"
})

-- Now you ONLY get chunks for this one object
-- Cursor advances automatically via orderedConsumer class
```

**Why specific subject works (without "all")**:

| Aspect | Wildcard (default) | Specific subject (default) |
|--------|-------------------|---------------------------|
| Scope | ALL chunks from ALL objects (unbounded) | Only chunks for ONE object (bounded) |
| Messages to read | Hundreds/thousands, keeps growing | Exactly N chunks (N = object chunk count) |
| Cursor behavior | Never reaches end | Reaches end, orderedConsumer handles resume |
| Memory usage | High (all objects) | Low (one object) |
| Bandwidth | Wastes on irrelevant data | Only relevant data |
| Protection | Wildcard + "all" BLOCKED by code | No restriction needed |

## Quick Reference: Which Approach for Which Use Case?

| Use Case | Approach | Consumer Type | Example |
|----------|----------|---------------|---------|
| Single `get("my-file")` | **Specific subject** | `filter_subject = "$O.bucket.C.<nuid>"` | Current `getOrderedConsumer()` |
| Bulk `getAll()` (50 objects) | **Wildcard** | `filter_subject = "$O.bucket.C.>"` | New `getAll()` method |
| Streaming export | **Wildcard + Subscribe** | Callback-based streaming | See "Approach 2" below |

**Key**: Wildcard is efficient for bulk operations, specific subject for single objects.

## deliverPolicy Options

| Policy | Behavior | Use Case | Protected | Lua |
|--------|----------|----------|-----------|-----|
| `all` | Deliver ALL from sequence 1 | First consumer creation, bounded data | YES - blocks wildcard and optStartSeq | YES |
| `last` | Start with last message received | Latest message only | No | YES |
| `new` | Only new messages after subscription | Watching for updates | No | YES |
| `by_start_sequence` | Start from opt_start_seq | Resuming from position | YES - blocks wildcard | YES |
| `by_start_time` | Start from opt_start_time timestamp | Time-based resume | No | YES |
| `last_per_subject` | Only last message per subject | Listing, metadata | No | YES |

**NATS server default**: `all` (when omitted)

**Lua implementation**:

- Default: `by_start_sequence` with `start_sequence=1` implied (efficient cursor tracking)
- Implemented: ALL 6 policies - `all`, `last`, `new`, `by_start_sequence`, `by_start_time`, `last_per_subject`
- **`start_sequence` takes priority**: When explicitly set, forces `deliver_policy="by_start_sequence"` regardless of default
- Wildcard + "all" **BLOCKED** (inefficient - reads unbounded data)
- Wildcard + optStartSeq **BLOCKED** (contradictory - unpredictable ordering)
- optStartTime + "all" **BLOCKED** (contradictory - "all" means from beginning)
- Specific subjects work fine without explicit deliverPolicy
- Use `start_time` or `opt_start_time` config for time-based delivery

**Why `by_start_sequence` as default**:

Using `by_start_sequence` with implied `start_sequence=1` is semantically equivalent to `deliver_policy="all"` but provides a unified mental model - you always start from some sequence number. This eliminates the conflict between "all" (quantity-based) and "start from X" (position-based) policies.

**Go's approach**: Uses `all` on first consumer creation (specific subject), then internally uses `by_start_sequence` for resets via cursor tracking. Lua simplifies this by always using the sequence-based approach.

## OrderedConsumer Cursor Behavior

The orderedConsumer class tracks cursor position internally:

```lua
-- First fetch: reads from sequence 1
local msgs = oc:fetch(5)  -- Gets sequences 1-5

-- Second fetch: continues from sequence 6
local msgs = oc:fetch(5)  -- Gets sequences 6-10

-- Cursor advances automatically!
-- No need to change deliver_policy
```

**Go's OrderedConsumer does this automatically**:

- Tracks `streamSeq` and `deliverSeq`
- On mismatch: creates new consumer with `opt_start_seq`
- No manual intervention needed

## When Would Wildcard Make Sense? For Bulk getAll() - Yes

We could use one wildcard ordered consumer to fetch all objects efficiently.

This makes sense when we need all objects anyway (like performance test fetching 50 objects).

### Implementation Status

**NOT implemented** - The wildcard approaches below are design proposals, not working code.

### What IS Implemented

From `nats-object-store.lua:189-224`, the `get == "all"` option:

```lua
-- Fetches ALL METADATA (not chunks)
if get == "all" then
    local consumerName = "object-store-get-" .. newNuid:sub(1, 8)
    local consumerCfg = {name = consumerName, filter_subject = ">"}  -- Wildcard!
    local cons = self.js:createOrUpdateConsumer(self.streamName, consumerCfg)

    -- Fetch all metadata messages
    for i = 1, cons.info.num_pending do
        obj = fromJson(messages:next().data)  -- Metadata only!
        arr[i] = obj
    end

    self.js:deleteConsumer(self.streamName, consumerName)  -- Deletes after use
    return {data = arr, info = cons.info}  -- Returns metadata array
end
```

This returns **metadata only** - no chunk data.

### Design Proposals (Not Implemented)

The following are design patterns for fetching complete objects (metadata + chunks) using wildcard consumers. These are **not currently implemented**.

#### Design Proposal 1: Wildcard + Fetch (Blocking, Stores All)

```lua
-- Current approach (per-object):
-- 50 get() calls = 50 consumer operations

-- Wildcard fetch approach (all-at-once) - NOT IMPLEMENTED:
-- 1. Fetch all metadata
local metaOc = js:orderedConsumer(streamName, {
    filter_subject = "$O.bucket.M.>",
    deliverPolicy = "last_per_subject"
})
local allMeta = metaOc:fetch(1000)

-- 2. Fetch ALL chunks with ONE wildcard consumer
local chunkOc = js:orderedConsumer(streamName, {
    filter_subject = "$O.bucket.C.>",
    deliverPolicy = "all"
})
local allChunks = chunkOc:fetch(10000)  -- BLOCKS until ALL fetched

-- 3. Group by nuid (stored in memory)
local objIdx = {}
for i = 1, #allChunks do
    local nuid = extractNuid(allChunks[i].subject)
    if objIdx[nuid] == nil then
        objIdx[nuid] = {}
    end
    objIdx[nuid][#objIdx[nuid] + 1] = allChunks[i].data
end

-- 4. Assemble each object
-- Result: ALL 50 objects with only 2 consumer operations!
```

#### Design Proposal 2: Wildcard + Subscribe (Streaming, Callback-based)

```lua
function objectStore:getAllChunkCount(callback)  -- NOT IMPLEMENTED
    -- First: get metadata to know expected chunk counts
    local metaOc = js:orderedConsumer(streamName, {
        filter_subject = "$O.bucket.M.>",
        deliverPolicy = "last_per_subject"
    })
    local metaMsgs = metaOc:fetch(1000)

    local chunkCountByNuid = {}
    for i = 1, #metaMsgs do
        local info = fromJson(metaMsgs[i].data)
        chunkCountByNuid[info.nuid] = info.chunks
    end

    -- Second: stream chunks with CALLBACK
    local chunkOc = js:orderedConsumer(streamName, {
        filter_subject = "$O.bucket.C.>",
        deliverPolicy = "all"
    })

    -- Stream chunks, group by nuid on-the-fly
    local objIdx = {}
    chunkOc:subscribe(function(msg)
        local nuid = extractNuid(msg.subject)

        if objIdx[nuid] == nil then
            objIdx[nuid] = {}
        end
        objIdx[nuid][#objIdx[nuid] + 1] = msg.data

        -- Check if object complete
        local expectedCount = chunkCountByNuid[nuid]
        if #objIdx[nuid] == expectedCount then
            -- Object complete! Return to caller immediately
            local data = table.concat(objIdx[nuid])
            callback(nuid, data)
            objIdx[nuid] = nil  -- Free memory immediately!
        end
    end)
end

-- Caller: processes objects as they arrive
bucket:getAllChunkCount(function(nuid, data)
    -- Each object callback as soon as complete
    -- No need to store all objects!
    print("Got object: " .. nuid .. ", size: " .. #data)
    processObject(data)  -- Caller handles storage
end)
```

### Comparison: Approaches for Bulk Fetch

| Aspect | Per-object fetch | Wildcard fetch (proposal) | Wildcard subscribe (proposal) |
|--------|-----------------|---------------------------|------------------------------|
| **Implemented?** | YES | NO | NO |
| Consumer operations | 50 | 2 | 2 |
| Total chunks fetched | 250 | 250 | 250 |
| Memory usage | One object | ALL objects | One object (free after callback) |
| First object latency | Immediate | After ALL fetches | **Immediate** (streaming) |
| Storage responsibility | Caller | All in memory | **Caller** (callback) |
| Complexity | Simple | Complex (grouping after) | Complex (on-the-fly grouping) |
| Best for | Single object | Small datasets | **Large datasets, streaming** |
| Go uses? | No | No | **Yes** |

### Key Insight: Subscribe + Callback = Best of Both Worlds

With subscribe + callback:

- **No need to store ALL objects** - free memory after each callback
- **Immediate first object** - callback fires as soon as chunks arrive
- **Caller controls storage** - can process, store, or discard
- **This is what Go does!** - `Subscribe()` with callback to `net.Pipe()`

### When to use each approach

**Wildcard + Subscribe** (streaming):

- Performance tests (fetching all objects anyway)
- Bulk export/backup (stream to file/network)
- Initial data load (stream to database)
- Batch processing (process each object)
- Large datasets (memory constrained)

**Wildcard + Fetch** (blocking):

- Small datasets (all objects fit in memory)
- Need random access to all objects
- Simple one-off scripts

**Per-object** (specific subject):

- Single `get("obj-1")` call
- Random access patterns
- Low latency needed on first object
- Need specific objects only

## getOrderedConsumer vs getDirect - What's the Difference?

### Current Implementation

| Method | Consumer Type | Caching | Deletion |
|--------|---------------|---------|----------|
| `getOrderedConsumer()` | Ephemeral ordered consumer (ord_1, ord_2...) | Cached by chunkSubject | Never deleted (only on bucket delete) |
| `getDirect()` | Durable named consumer (object-store-direct-abc123) | Cached by chunkSubject | Never deleted (only on bucket delete) |

### Key Differences

| Aspect | `getOrderedConsumer()` | `getDirect()` |
|--------|------------------------|---------------|
| **Consumer class** | `orderedConsumer` class | `pullConsumer` class |
| **Consumer name** | Auto-generated (ord_1, ord_2...) | Random but persistent (object-store-direct-xxx) |
| **Auto-reset** | YES - on sequence mismatch | NO - manual handling required |
| **deliver_policy** | User-configurable (default: by_start_sequence) | Fixed: instant replay |
| **ack_policy** | none | none |
| **Primary use** | General object fetching | Default path for `get()` |

### Why Have Two Methods?

**Original design intent**:

- `getOrderedConsumer()`: Sophisticated fetching with auto-reset on errors
- `getDirect()`: Simple, lightweight fetching for simple cases

**Current reality**:

- Both cache consumers identically
- Both do NOT delete after use
- `getDirect()` is the DEFAULT (via `use_direct_get ~= false`)
- The main difference is in error handling complexity

## Consumer Behavior Analysis

### Actual Code Behavior (From Source)

The code at `nats-object-store.lua:319-390` ALREADY implements what was previously recommended:

```lua
-- Line 342: CHECKS CACHE
local cons = self.directConsumerCache[chunkSubject]

if cons == nil then
    -- Create consumer...
    self.directConsumerCache[chunkSubject] = cons  -- Line 351: CACHES IT
end

-- NO deleteConsumer call here!
```

**Consumers are ONLY deleted on**:

- `objectStore:delete()` - lines 419-422
- `objectStore:watch()` - similar pattern
- Bucket delete/destroy

### Documentation Was Incorrect

The previous documentation claimed:

- "No cache" ❌ **WRONG** - Code DOES cache (line 342, 351)
- "Deleted after each use" ❌ **WRONG** - Code NEVER deletes after use
- "Creates random consumer name each time" ❌ **MISLEADING** - Random name but cached by chunkSubject

The documentation was based on outdated assumptions or a different implementation.

### Current Implementation is CORRECT

The current code already follows best practices:

1. ✅ Caches consumers by chunkSubject
2. ✅ Never deletes consumers (only on bucket purge)
3. ✅ Uses specific subjects (not wildcard) for single-object get
4. ✅ Both `getDirect()` and `getOrderedConsumer()` cache identically

### Actual Delete Behavior

From `nats-object-store.lua:392-424`:

```lua
function objectStore:delete(name)
    -- ... delete logic ...

    -- Clear ordered consumer cache and delete consumers
    for _, oc in pairs(self.orderedConsumerCache) do
        deleteOrderedConsumer(self.js, self.streamName, oc)
    end
    self.orderedConsumerCache = {}

    -- Clear direct consumer cache and delete consumers
    for _, cons in pairs(self.directConsumerCache) do
        deleteDirectConsumer(self.js, self.streamName, cons)
    end
    self.directConsumerCache = {}

    return purgeResult
end
```

Consumers are ONLY deleted when:

- An object is deleted from the bucket
- The bucket is sealed/purged/destroyed

## Connection Stress Issues

### Previous Concern (INCORRECT)

The previous documentation claimed:

- "Rapid consumer create/delete operations stress NATS server"
- "50 objects × (1 create + 1 delete) = 100 operations"

**This was based on incorrect assumptions.**

### Actual Behavior (CORRECT)

The current implementation:

- Creates ONE consumer per unique chunkSubject
- Caches consumers for reuse
- NEVER deletes consumers during normal operations
- Only deletes on bucket/object delete

For 50 objects accessed repeatedly:

- 50 consumer creations (one per unique object)
- ZERO deletions during normal operation
- Consumers reused on subsequent access

This is the CORRECT pattern - minimal server load.

## Direct GET API Usage

### Available APIs in nats-api.lua

| API Constant | Template | Used? | Purpose |
|--------------|----------|-------|---------|
| `apiDirectMsgGetT` | `$JS.API.DIRECT.GET.%s` | **NO** | Get by sequence (defined but unused) |
| `apiDirectMsgGetLastBySubjectT` | `$JS.API.DIRECT.GET.%s.%s` | **YES** | Get last by subject (metadata) |
| `apiMsgGetT` | `$JS.API.STREAM.MSG.GET.%s` | **YES** | Get via stream API |

### How Metadata is Retrieved (TRUE Direct GET)

From `nats-jetstream.lua:103-136`:

```lua
function streamImpl:getLastMsgForSubject(msgSubject)
    if self.config.allow_direct then
        -- TRUE direct GET - no consumer needed!
        subject = format(api.apiDirectMsgGetLastBySubjectT, self.name, msgSubject)
        local resp = self.js:apiRequest(subject, nil)
        -- Returns ONE message directly
    end
    -- Fallback to stream API...
end
```

**This is TRUE direct GET** - bypasses consumers entirely.

### Why Not Use Direct GET for Chunks?

**Problem**: We don't have sequence numbers for chunks!

- Metadata retrieval: We have the subject (`$O.bucket.M.name`) → Direct GET works
- Chunk retrieval: We have the nuid but NOT sequence numbers → Must use consumer

**Alternative**: Loop-based direct GET (one request per chunk) = SLOW for multi-chunk objects

The current consumer-based approach is more efficient for fetching multiple chunks.

## Summary

1. **Metadata retrieval** = TRUE direct GET (`apiDirectMsgGetLastBySubjectT`) - no consumer
2. **Chunk retrieval** = Consumer-based (both `getDirect()` and `getOrderedConsumer()`) - requires consumer
3. **Both methods cache consumers** identically - no difference in performance
4. **NO wildcard for single-object get** - filter by specific chunk subject with nuid
5. **deliver_policy = "all"** is OK for specific subjects (bounded), BAD for wildcards (unbounded)
6. **Current implementation is CORRECT** - follows best practices
7. **`useDirectGet` option** controls which consumer type to use, not whether to use direct GET API
8. **`get == "all"` option exists** but only fetches metadata, not complete objects
9. **Wildcard for complete objects (chunks) is NOT implemented** - only design proposals exist
