# NATS Grid Messaging System

## Overview

The NATS Grid Messaging System provides a complete solution for transmitting large grid datasets, real-time change notifications, chat messaging, and presence tracking using NATS JetStream technology.

## Features

- **Message-based Grid Transmission**: Send large datasets in messages (100 rows initial, 1000 rows subsequent)
- **Ordered Consumer**: Guaranteed message delivery order for reliable consumption
- **Change Messages**: Incremental updates using new/change/delete patterns
- **Client-to-Client Chat**: Direct messaging and broadcast support
- **Presence Tracking**: Online/offline/away status with automatic timeout cleanup

## Architecture

### Server-Side (Lua)

```text
lib/message/nats/
├── nats-grid-config.lua   -- Configuration constants
├── nats-grid.lua          -- Core grid messaging module
└── nats.lua               -- NATS connection API
```

### Client-Side (JavaScript)

```text
src/core/
└── nats-grid.js             -- Message handling, chat, presence
```

## Configuration

### File: `nats-grid-config.lua`

```lua
local gridConfig = {
	-- Message counts for grid transmission
	first_message_count = 100,     -- First message: max 100 rows
	message_count = 1000,          -- Subsequent messages: max 1000 rows

	-- Message types
	message_type_grid = "grid",
	message_type_chat = "chat",
	message_type_presence = "presence",
	message_type_change = "change",

	-- Subject prefixes
	subject_grid_prefix = "grid.",
	subject_chat_prefix = "chat.",
	subject_presence_prefix = "presence.",
	subject_change_prefix = "change.",

	-- Ordered consumer settings
	ordered_consumer_name_prefix = "ord",
	consumer_inactive_threshold = 5 * 60 * 1e9, -- 5 minutes

	-- Stream configuration
	stream_storage = "file",
	stream_num_replicas = 1,
	stream_max_age = 24 * 60 * 60 * 1e9,

	-- Consumer configuration
	consumer_ack_policy = "none",
	consumer_deliver_policy = "all",
	consumer_replay_policy = "instant",
	consumer_flow_control = true,
	consumer_idle_heartbeat = 5 * 1e9,
}
```

## Message Formats

### Grid Message

```json
{
	"message_type": "grid",
	"message_id": "unique-id",
	"bucket": "bucket-name",
	"msg_index": 0,
	"message_count": 3,
	"row_count": 100,
	"data": [
		{"record_id": "id1", "field1": "value1"},
		{"record_id": "id2", "field1": "value2"}
	],
	"sequence": 1,
	"timestamp": "2024-01-01T00:00:00Z"
}
```

### Change Message

```json
{
	"message_type": "change",
	"message_id": "unique-id",
	"bucket": "bucket-name",
	"change_arr": {
		"table_name": [
			{
				"record_id": "id1",
				"__change": "new",
				"field1": "value1"
			},
			{
				"record_id": "id2",
				"__change": "change",
				"field1": "updated"
			},
			{
				"record_id": "id3",
				"__change": "delete"
			}
		]
	},
	"sequence": 123,
	"timestamp": "2024-01-01T00:00:00Z"
}
```

### Chat Message

```json
{
	"message_type": "chat",
	"message_id": "unique-id",
	"from_client": "client1",
	"to_client": "client2",
	"message": "Hello!",
	"timestamp": "2024-01-01T00:00:00Z"
}
```

Broadcast chat uses `"to_client": "*"`.

### Presence Message

```json
{
	"message_type": "presence",
	"message_id": "unique-id",
	"client_id": "client1",
	"status": "online",
	"timestamp": "2024-01-01T00:00:00Z"
}
```

Valid statuses: `"online"`, `"offline"`, `"away"`.

## Server-Side Usage (Lua)

### Sending Grid Data

```lua
local grid = require "message.nats.nats-grid"

-- Create shared configuration object
local messageOption = {
	first_message_count = 100,
	message_count = 1000
}

-- Send grid with configuration
local sent = grid.sendGrid("bucket-name", gridData, messageOption)

-- Returns:
-- {row_count = 250, message_count = 3}

-- ALWAYS store the result for use with receiveGrid
local fetched = grid.receiveGrid("bucket-name", sent)
```

### Helper Function for Expected Message Count

```lua
-- Calculate expected message count from row count
local function getExpectedMessageCount(rowCount, option)
	if rowCount <= option.first_message_count then
		return 1
	end
	return 1 + math.ceil((rowCount - option.first_message_count) / option.message_count)
end

-- Usage in tests
local rowCount = 250
local sent = grid.sendGrid("bucket", gridData, messageOption)
test.equal(sent.message_count, getExpectedMessageCount(rowCount, messageOption))
```

### Sending Change Messages

```lua
local changeArr = {
	table_name = {
		{record_id = "id1", field1 = "new", __change = "new"},
		{record_id = "id2", field1 = "mod", __change = "change"},
		{record_id = "id3", __change = "delete"}
	}
}

local ret = grid.sendChangeMessage("bucket-name", changeArr)
```

### Change Detection

```lua
-- Automatically detect changes between old and new data
local ret = grid.sendGridWithChanges("bucket-name", oldGridData, newGridData)
```

### Chat Messages

```lua
-- Direct message
local ret = grid.sendChatMessage("from-client", "to-client", "Hello!")

-- Broadcast to all
local ret = grid.broadcastChatMessage("from-client", "Broadcast message")
```

### Presence Updates

```lua
local ret = grid.sendPresence("client-id", "online")
local ret = grid.sendPresence("client-id", "away")
local ret = grid.sendPresence("client-id", "offline")
```

### Ordered Consumer

```lua
-- Create ordered consumer for reliable message consumption
local consumer = grid.createOrderedConsumer("bucket-name")

-- Fetch messages
local msgs = consumer:fetch(10)

-- Iterate through messages
local iter = consumer:messages({pull_max_message_count = 100})
local msg = iter:next()
```

### Receiving Grid Data

```lua
-- ALWAYS store sendGrid result in a named variable
local sent = grid.sendGrid("bucket-name", gridData, messageOption)

-- Pass the ENTIRE sendGrid return object to receiveGrid
-- sent object acts as the option parameter, containing message_count and optional start_sequence
local fetched = grid.receiveGrid("bucket-name", sent)

-- sent contains: {row_count = N, message_count = M, [start_sequence = 1]}
-- receiveGrid uses sent.message_count internally to fetch messages
-- fetched is an array of messages with length equal to sent.message_count

-- Array decoding: create new array from messages
sent.array = true
local fetched = grid.receiveGrid("bucket-name", sent)
local array = fetched.array  -- Array of decoded message objects

-- Array accumulation: append to existing array
sent.array = existingArray
local fetched = grid.receiveGrid("bucket-name", sent)
local totalCount = fetched.array_total_count  -- Total count after accumulation

-- Force reading from beginning (useful for re-reading after initial consumption)
sent.start_sequence = 1
local fetched = grid.receiveGrid("bucket-name", sent)
```

## Client-Side Usage (JavaScript)

### Subscribing to Messages

```javascript
import { connectNats, subscribeToGrid } from '/src/core/nats-grid.js'

const nats = await connectNats()

// Subscribe to grid updates
const gridSub = subscribeToGrid(nats, 'bucket-name', (msg) => {
	console.log('Received grid:', msg.row_count, 'rows')
	// msg.data contains the assembled grid data
})

// Subscribe to change messages
const changeSub = subscribeToChange(nats, 'bucket-name', (msg) => {
	console.log('Changes:', msg.change_arr)
	// Process changes using change-data.js patterns
})

// Subscribe to chat
const chatSub = subscribeToChat(nats, 'client-id', (msg) => {
	console.log('Chat from', msg.from_client, ':', msg.message)
})

// Subscribe to presence
const presenceSub = subscribeToPresence(nats, 'client-id', (msg) => {
	console.log('Client', msg.client_id, 'is', msg.status)
})
```

### Publishing Messages

```javascript
import { sendGrid, sendChangeMessage, sendChatMessage, sendPresence } from '/src/core/nats-grid.js'

// Create shared configuration object
const messageOption = {
	first_message_count: 100,
	message_count: 1000
}

// Send grid data with configuration
const gridRet = sendGrid('bucket-name', gridData, messageOption)

// Returns object with: {row_count, message_count}
// This object IS the option parameter for receiveGrid

// Decode messages to array
import { receiveGrid } from '/src/core/nats-grid.js'
gridRet.array = true  // Set array option directly on the result
const received = receiveGrid('bucket-name', gridRet)
const array = received.array  // Array of message objects

// Accumulate messages in existing array
gridRet.array = existingArray  // Pass existing array
const accumulated = receiveGrid('bucket-name', gridRet)
const totalCount = accumulated.array_total_count  // Total count after accumulation

// Force reading from beginning
gridRet.start_sequence = 1
const refetched = receiveGrid('bucket-name', gridRet)

// Send changes
const changeRet = sendChangeMessage('bucket-name', changeArr)

// Send chat message
const chatRet = sendChatMessage('from-client', 'to-client', 'Hello!')

// Update presence
const presenceRet = sendPresence('client-id', 'online')
```

### Message Processing

The client automatically:

- **Buffers messages** as they arrive
- **Assembles complete grids** when all messages are received
- **Processes change messages** using `change-data.js` patterns
- **Tracks presence** with automatic cleanup of stale entries (10 minutes)
- **Maintains chat history** (limit to 100 messages)

### Reactive State

```javascript
import { messageBuffer, presenceList, chatMessages } from '/src/core/nats-grid.js'

// Watch for assembled grids
watch(messageBuffer, (buffer) => {
	for (const msgId in buffer) {
		const grid = assembleGrid(buffer[msgId])
		// Process complete grid
	}
})

// Watch presence updates
watch(presenceList, (list) => {
	for (const clientId in list) {
		console.log(clientId, list[clientId].status)
	}
})

// Watch chat messages
watch(chatMessages, (msgs) => {
	msgs.forEach(msg => console.log(msg.from_client, msg.message))
})
```

## Testing

### Test Pattern (Lua)

Proper test structure with shared configuration and helper function:

```lua
local gridTest = {
	messageOption = {first_message_count = 100, message_count = 1000},
	smallRowCount = 50,
	normalRowCount = 250,
	basicSent = nil,
	multiSent = nil
}

local function getExpectedMessageCount(rowCount)
	if rowCount <= gridTest.messageOption.first_message_count then
		return 1
	end
	return 1 + math.ceil((rowCount - gridTest.messageOption.first_message_count) / gridTest.messageOption.message_count)
end

function test.nats.grid.basic()
	local bucket = "basic-test"
	local gridData = createTestGrid(gridTest.smallRowCount)
	gridTest.basicSent = grid.sendGrid(bucket, gridData, gridTest.messageOption)
	test.equal(gridTest.basicSent.message_count, getExpectedMessageCount(gridTest.smallRowCount))
end

function test.nats.read.basic()
	local bucket = "basic-test"
	local fetched = grid.receiveGrid(bucket, gridTest.basicSent)
	test.equal(#fetched, gridTest.basicSent.message_count)
end

-- For re-reading from beginning (after previous consumption)
function test.nats.read.again()
	local bucket = "basic-test"
	gridTest.basicSent.start_sequence = 1  -- Force read from sequence 1
	local fetched = grid.receiveGrid(bucket, gridTest.basicSent)
	test.equal(#fetched, gridTest.basicSent.message_count)
end
```

**Key Rules:**

1. ALL row counts in `gridTest` table (file-scope)
2. ALL `grid.sendGrid` results stored in `gridTest.*Sent`
3. ALWAYS use 3rd parameter (messageOption)
4. EVERY `grid.sendGrid` MUST have corresponding read test
5. NO inline option tables - use shared `messageOption`

### Lua Tests

Run comprehensive NATS grid tests:

```bash
# Run all NATS tests
lj /Volumes/nc/nc-server/test/lib/nats-test.lua

# Run specific grid tests with trace
lj /Volumes/nc/nc-server/test/lib/nats-test.lua -t

# Run dedicated grid tests
lj /Volumes/nc/nc-server/test/lib/nats-grid-test.lua

# Stress test with configurable row count
lj /Volumes/nc/nc-server/test/lib/nats-grid-test.lua stress 100000
```

### JavaScript Tests

Run Vitest tests for client-side grid handling:

```bash
# Run all NATS grid tests
bun test /Volumes/nc/nc-frontend/test/nats-grid.test.js

# Run with watch mode
bun test --watch /Volumes/nc/nc-frontend/test/nats-grid.test.js
```

## Performance

### Message Strategy

- **First message**: 100 rows (optimized for quick initial display)
- **Subsequent messages**: 1000 rows (optimized for throughput)

### Examples

| Rows | First Message | Subsequent Messages | Total Messages |
|------|---------------|---------------------|----------------|
| 50   | 1 (50 rows)   | 0                   | 1              |
| 250  | 1 (100 rows)  | 1 (150 rows)        | 2              |
| 1500 | 1 (100 rows)  | 2 (1400 rows)       | 3              |
| 5250 | 1 (100 rows)  | 6 (5150 rows)       | 7              |

### Timing

Typical performance on localhost:

- **1000 rows**: ~0.1 seconds
- **10,000 rows**: ~0.5 seconds
- **100,000 rows**: ~5 seconds

## Stream Management

### Stream Naming

Grid streams use the pattern: `GRID_<bucket>`

Example: `GRID_work-center`

### Stream Lifecycle

- Streams are automatically created on first message
- Maximum age: 24 hours (configurable via `stream_max_age`)
- Automatic cleanup of old messages

## Subject Hierarchy

```text
grid.<bucket>         -- Grid data messages
change.<bucket>       -- Change messages
chat.<client-id>      -- Client-to-client chat
chat.*                -- Broadcast chat
presence.<client-id>  -- Presence updates
```

## Error Handling

### Connection Errors

```lua
local ret = grid.sendGrid("bucket", gridData)
if ret.error then
	-- Handle error
	util.printError("Grid send failed: %s", ret.error)
end
```

### Consumer Errors

```lua
local consumer = grid.createOrderedConsumer("bucket")
if consumer.error then
	-- Consumer creation failed
	util.printError("Consumer failed: %s", consumer.error)
end
```

### Client-Side Errors

```javascript
try {
	const nats = await connectNats()
} catch (err) {
	console.error('NATS connection failed:', err)
}
```

## Best Practices

1. **Always use ordered consumers** for reliable message ordering
2. **Enable change detection** for incremental updates instead of full grid transmission
3. **Use presence tracking** to monitor client connectivity
4. **Implement cleanup** for stale message buffers (client-side)
5. **Configure message counts** based on network conditions
6. **Monitor stream sizes** to prevent excessive storage

## API Reference

### grid.sendGrid(bucket, gridData, option)

Sends grid data as multiple messages.

**Parameters:**

- `bucket` (string): NATS stream bucket name
- `gridData` (table): Grid data with `nats_name` and `data` array
- `option` (table): Optional parameters
  - `first_message_count` (number): Rows in first message (default: 100)
  - `message_count` (number): Rows in subsequent messages (default: 1000)

**Returns:**

```lua
{
	row_count = 250,      -- Total rows sent
	message_count = 3     -- Total messages sent
}
```

### grid.receiveGrid(bucket, option)

Receives grid messages using the return value from `sendGrid`.

**Parameters:**

- `bucket` (string): NATS stream bucket name
- `option` (table): Return value from `sendGrid` containing:
  - `message_count` (number): Required - number of messages to fetch
  - `row_count` (number): For display/logging purposes
  - `start_sequence` (number, optional): Force starting from specific sequence (default: use cursor)
  - `array` (boolean/table): If true, returns decoded array; if table, appends to existing array

**Returns:**
Array of received messages. If `array` option is set, includes `array` property with decoded data.

### grid.sendChangeMessage(bucket, changeArr, option)

Sends change message with new/change/delete records.

### grid.sendGridWithChanges(bucket, oldGridData, newGridData, option)

Detects changes and sends them as a change message.

### grid.sendChatMessage(fromClient, toClient, messageText, option)

Sends direct chat message between clients.

### grid.broadcastChatMessage(fromClient, messageText, option)

Broadcasts chat message to all clients.

### grid.sendPresence(clientId, status, option)

Sends presence status update.

## Troubleshooting

### Messages Not Arriving

1. Check stream exists: `js:streamInfo("GRID_<bucket>")`
2. Verify consumer is created: `js:consumerInfo("GRID_<bucket>", "consumer-name")`
3. Check network connectivity to NATS server

### Message Assembly Issues

1. Verify `message_id` is consistent across messages
2. Check `msg_index` and `message_count` match
3. Enable logging for message reception debugging

### Performance Issues

1. Reduce message counts for slower networks
2. Use change messages instead of full grid transmission
3. Consider stream compression

## References

- [NATS JetStream Documentation](https://docs.nats.io/nats-concepts/jetstream)
- [Lua NATS Client](lib/message/nats/nats.lua)
- [JavaScript NATS Client](https://github.com/nats-io/nats.ws)

## Changelog

### 2025-01-30

- **Breaking**: `receiveGrid()` signature changed from `receiveGrid(bucket, sentObject, option)` to `receiveGrid(bucket, option)`
- The `sentObject` from `sendGrid()` IS the option parameter - contains `message_count`, `row_count`, and optional `start_sequence`
- Added `start_sequence` option support - when set, forces reading from specific sequence (use `start_sequence=1` to re-read from beginning)
- Updated ordered consumer to prioritize explicit `start_sequence` over cursor state (explicit > implicit)
- When `start_sequence` is explicitly set, forces `deliver_policy="by_start_sequence"` to avoid conflicts with default `all` policy
- Simplified test structure: send tests no longer consume messages, allowing read tests to properly verify consumption

### 2025-01-29

- Updated all tests to use shared `gridTest` configuration object
- Added `getExpectedMessageCount()` helper function to eliminate hardcoded expected counts
- EVERY `grid.sendGrid()` result now stored in `gridTest.*Sent` (no more `ret =` or local variables)
- EVERY `grid.sendGrid()` call now uses explicit 3rd parameter `messageOption`
- EVERY `grid.sendGrid()` result has corresponding read test for complete coverage
- Removed all inline option tables in favor of shared configuration objects

### 2024-01-29

- Renamed parameters: `initial_chunk_size` → `first_message_count`, `subsequent_chunk_size` → `message_count`
- Changed return values: `total_chunks` → `message_count`, `chunk_index` → `msg_index`, `total_chunks` → `message_count`
- Updated `receiveGrid` to accept full `sentObject` return value from `sendGrid`

### 2024-01-25

- Initial implementation of NATS Grid Messaging System
- Message-based grid transmission (100/1000 row messages)
- Change message support (new/change/delete)
- Client-to-client chat and presence tracking
- Ordered consumer support
- Comprehensive testing (Lua and Vitest)
