# iComposer Message Management User Guide

## Overview

iComposer Message Management is a distributed transactional messaging framework that enables asynchronous, reliable message delivery and execution within iComposer orchestration flows. Key capabilities include:

- Asynchronous message scheduling and execution
- Automatic retry and failure recovery
- Distributed locking to ensure single-node processing in clustered environments
- TCC transaction support (Seata integration)
- Automatic archiving of completed messages to a history table
- REST API for message querying, manual retry, and status reset

## Architecture Overview

```
┌──────────────┐     TCC prepare      ┌──────────────────────┐
│  Business    │ ─────────────────►   │  T_ICMP_TX_MESSAGE   │  (current table)
│  Code        │                      │  status=NON_COMMIT(-1)│
│ (Groovy API) │                      └──────────┬───────────┘
└──────────────┘                                 │ TCC commit → status=INITIAL(0)
                                                 ▼
                                      ┌──────────────────────┐
                                      │ IComposerMessage      │
                                      │ Scheduler (polling)   │
                                      └──────────┬───────────┘
                                                 │ mark DOING(1)
                                                 ▼
                                      ┌──────────────────────┐
                                      │ Thread Pool           │
                                      │ GroovyRestfulExecutor │
                                      └──────────┬───────────┘
                                           ┌─────┴──────┐
                                           ▼            ▼
                                      SUCCESS(2)    ERROR(3)
                                           │            │
                                           │       auto-retry (≤maxRetryTimes)
                                           ▼            ▼
                                      ┌──────────────────────────┐
                                      │ T_ICMP_TX_MESSAGE_HIS    │  (history table)
                                      └──────────────────────────┘
```

## Message Status Lifecycle

| Status Code | Name | Description |
|-------------|------|-------------|
| `-1` | NON_COMMIT | TCC prepare phase; message reserved but not yet committed |
| `0` | INITIAL | Committed; waiting for scheduler to pick up and execute |
| `1` | DOING | Currently being processed |
| `2` | SUCCESS | Executed successfully |
| `3` | ERROR | Execution failed; awaiting retry |
| `4` | BLOCK | Blocked; requires manual intervention |

Status transition paths:

```
NON_COMMIT(-1) ──TCC commit──► INITIAL(0) ──scheduler──► DOING(1) ──success──► SUCCESS(2)
                                                      ──fail───► ERROR(3) ──retry──► DOING(1)
                                                                 ──maxRetry──► BLOCK(4)
NON_COMMIT(-1) ──TCC rollback──► deleted
DOING(1) ──manual reset──► INITIAL(0)
```

## Configuration

Configure the following parameters in `application.properties` or via environment variables:

| Parameter | Default | Description |
|-----------|---------|-------------|
| `icomposer.message.enabled` | None (must be explicitly set to `true`) | Enable the message scheduler |
| `icomposer.message.cron` | `0/5 * * * * ?` | Cron expression for message processing (default: every 5 seconds) |
| `icomposer.message.backupCron` | `0/10 * * * * ?` | Cron expression for message archiving (default: every 10 seconds) |
| `icomposer.message.batchSize` | `10` | Number of messages fetched per scheduling cycle |
| `icomposer.message.schedulerLockTimeoutSeconds` | `30` | Distributed lock timeout in seconds |
| `icomposer.message.retryIntervalMinutes` | `2` | Retry interval for failed messages in minutes |
| `icomposer.message.maxRetryTimes` | `5` | Maximum number of retry attempts |
| `icomposer.message.backupBatchSize` | `500` | Batch size for archiving messages |
| `icomposer.message.useSystemUser` | `false` | Use system user for execution (`false` = use message creator) |

### Minimal Configuration

```properties
# Enable message scheduler
icomposer.message.enabled=true
```

### Production Tuning Example

```properties
icomposer.message.enabled=true
icomposer.message.cron=0/3 * * * * ?
icomposer.message.batchSize=20
icomposer.message.retryIntervalMinutes=5
icomposer.message.maxRetryTimes=10
icomposer.message.backupBatchSize=1000
icomposer.message.useSystemUser=false
```

## REST API

All API paths are prefixed with `/v1/flow/message`.

### 1. Query Current Messages (Paginated)

**POST** `/v1/flow/message/pagedQueryCurrent`

Query messages from the current message table (`T_ICMP_TX_MESSAGE`).

**Request Body:**

```json
{
  "pageIndex": 1,
  "pageSize": 20,
  "businessNo": "POL-2025-001",
  "functionName": "policyService",
  "methodName": "issuePolicy",
  "globalTransactionId": "tx-uuid-xxx",
  "status": "3",
  "traceId": "trace-xxx",
  "submitTimeFrom": "2025-05-01T00:00:00",
  "submitTimeTo": "2025-05-31T23:59:59",
  "processTimeFrom": null,
  "processTimeTo": null
}
```

**Query Parameters:**

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| `pageIndex` | Integer | Yes | Page number (1-based) |
| `pageSize` | Integer | Yes | Page size |
| `businessNo` | String | No | Business number (fuzzy match) |
| `functionName` | String | No | Groovy function name |
| `methodName` | String | No | Groovy method name |
| `globalTransactionId` | String | No | Global transaction ID |
| `status` | String | No | Message status code |
| `traceId` | String | No | Trace ID |
| `submitTimeFrom/To` | LocalDateTime | No | Submit time range |
| `processTimeFrom/To` | LocalDateTime | No | Process time range |
| `endTimeFrom/To` | LocalDateTime | No | End time range |

**Response:**

```json
{
  "data": [
    {
      "messageId": 1001,
      "globalTransactionId": "tx-uuid-xxx",
      "traceId": "trace-xxx",
      "businessNo": "POL-2025-001",
      "submitTime": "2025-05-20T10:30:00",
      "processTime": "2025-05-20T10:30:05",
      "endTime": null,
      "transactionType": 1,
      "isSuccessful": "N",
      "errorMessage": "Connection timeout",
      "functionName": "policyService",
      "methodName": "issuePolicy",
      "status": "3",
      "redoTimes": 2,
      "lastRedoTime": "2025-05-20T10:35:05"
    }
  ],
  "totalCount": 150,
  "pageIndex": 1,
  "pageSize": 20
}
```

### 2. Query Historical Messages (Paginated)

**POST** `/v1/flow/message/pagedQuery`

Query messages from the history archive table (`T_ICMP_TX_MESSAGE_HIS`). The request body and response format are identical to `pagedQueryCurrent`.

### 3. Retry Failed Messages

**POST** `/v1/flow/message/redo`

Re-queue failed messages from the history table back into the current message table for re-execution.

**Request Body:**

```json
[1001, 1002, 1003]
```

A list of `hisId` values from the history table.

**Response:**

```json
{
  "code": "0",
  "message": "success"
}
```

> **Note:** The retry operation extracts message details from the history table, re-inserts them into the current message table with INITIAL(0) status, and the scheduler picks them up automatically.

### 4. Reset Message Status

**POST** `/v1/flow/message/reset`

Reset DOING(1) messages back to INITIAL(0). Used to recover messages stuck in DOING state due to node crashes or other failures.

**Request Body:**

```json
[2001, 2002]
```

A list of `messageId` values from the current message table.

**Response:**

```json
{
  "code": "0",
  "message": "success"
}
```

> **Use case:** When messages remain in DOING status for an extended period (e.g., the executing node crashed), use this API to reset them to INITIAL so the scheduler can pick them up again.

## Usage in Groovy Code

Use `IComposerExecutorUtils.asyncExecuteWithMessage` to submit asynchronous messages. This method internally uses the TCC pattern to ensure consistency between the message and the business transaction: the message takes effect when the outer transaction commits, and is automatically deleted on rollback.

### Basic Usage

```groovy
import com.insuremo.icomposer.utils.IComposerExecutorUtils

// Minimal form: businessNo + function + method + params
IComposerExecutorUtils.asyncExecuteWithMessage(
    "POL-2025-001",       // businessNo - business number (for querying and tracking)
    "policyService",      // serviceFunction - Groovy function name to invoke
    "issuePolicy",        // methodName - method name to execute
    policyData            // params - method parameters (varargs)
)
```

### Specifying Transaction Type

```groovy
// With custom txType (transaction type identifier, defaults to 100)
IComposerExecutorUtils.asyncExecuteWithMessage(
    "POL-2025-001",       // businessNo
    200L,                 // txType - custom transaction type
    "policyService",      // serviceFunction
    "issuePolicy",        // methodName
    policyData            // params
)
```

### Using Map Parameters

When there are many parameters, you can pass them as a `Map` (use `LinkedHashMap` to preserve parameter order):

```groovy
// Map parameter form
IComposerExecutorUtils.asyncExecuteWithMessage(
    "POL-2025-001",       // businessNo
    "policyService",      // serviceFunction
    "issuePolicy",        // methodName
    [
        policyNo: "POL-2025-001",
        productCode: "TRAVEL-PA",
        insuredAmount: 100000
    ]                     // methodParamsMap - LinkedHashMap preserves order
)

// Map parameters + custom txType
IComposerExecutorUtils.asyncExecuteWithMessage(
    "POL-2025-001",       // businessNo
    200L,                 // txType
    "policyService",      // serviceFunction
    "issuePolicy",        // methodName
    [policyNo: "POL-2025-001", productCode: "TRAVEL-PA"]
)
```

### Method Signatures

| Signature | Description |
|-----------|-------------|
| `asyncExecuteWithMessage(businessNo, serviceFunction, methodName, Object... params)` | Basic form; txType defaults to 100 |
| `asyncExecuteWithMessage(businessNo, txType, serviceFunction, methodName, Object... params)` | Specify transaction type |
| `asyncExecuteWithMessage(businessNo, serviceFunction, methodName, Map<String, Object> paramsMap)` | Map parameters; txType defaults to 100 |
| `asyncExecuteWithMessage(businessNo, txType, serviceFunction, methodName, Map<String, Object> paramsMap)` | Map parameters + custom transaction type |

### Tracing Message Execution

A `traceId` is automatically assigned during message execution. Use the traceId in your logging system to trace the full execution chain.

## How the Scheduler Works

### Message Processing Flow

1. **Scheduled polling**: The scheduler triggers based on the cron configuration (default: every 5 seconds)
2. **Distributed lock**: Acquires a non-reentrant lock to ensure only one node processes messages at a time
3. **Fetch pending messages**:
   - Prioritizes INITIAL(0) status messages
   - If fewer than `batchSize`, supplements with ERROR(3) messages eligible for retry
4. **Mark as executing**: Uses optimistic locking to change status from INITIAL/ERROR to DOING
5. **Thread pool execution**: Submits messages to the thread pool for asynchronous Groovy function execution
6. **Record result**: Updates status to SUCCESS or ERROR after execution, recording traceId and error details

### Message Archiving Flow

1. The scheduler triggers based on `backupCron` (default: every 10 seconds)
2. Queries completed messages (SUCCESS and ERROR/BLOCK that have reached the retry limit)
3. Batch-migrates them to the history table

## Operations Guide

### Common Scenarios

#### View Failed Messages

```
POST /v1/flow/message/pagedQueryCurrent
{
  "pageIndex": 1,
  "pageSize": 50,
  "status": "3"
}
```

#### Manually Retry Failed Messages

1. Find the `hisId` of messages to retry from the history table
2. Call the redo API:
```
POST /v1/flow/message/redo
[1001, 1002]
```

#### Reset Stuck DOING Messages

When messages remain in DOING status due to a node crash:

```
POST /v1/flow/message/reset
[2001, 2002]
```

#### Track Messages by Business Number

```
POST /v1/flow/message/pagedQueryCurrent
{
  "pageIndex": 1,
  "pageSize": 20,
  "businessNo": "POL-2025-001"
}
```

### Monitoring Checklist

- Watch for accumulation of ERROR and BLOCK status messages
- Monitor thread pool queue size and active thread count
- Watch for `taskExecutor rejected` warnings in scheduler logs
- Check if distributed locks are timing out frequently

## Important Notes

- The scheduler is only activated when `icomposer.message.enabled=true`
- Distributed locking is database-based, relying on the `LockUtils` utility
- Message execution uses the message creator's user context; falls back to ADMIN user if the user no longer exists
- When `icomposer.message.useSystemUser=true`, all messages execute under the system user identity
- Messages that reach `maxRetryTimes` are not automatically archived and require manual handling via the redo/reset APIs