[Feature Request] Storage as Queue

# Asynchronous Inventory Processing Implementation

## Issue Type
Enhancement

## Priority
High

## Summary
Implement asynchronous inventory processing architecture to improve system performance and scalability by decoupling inventory availability calculations from transactional documents processing.

---

## Background

### Problem Statement
The current ADempiere inventory management system processes stock calculations synchronously during document completion (Sales Orders, Purchase Orders, Shipments, Production Orders, etc.). This approach causes:

1. **Performance bottlenecks** during high-volume transaction periods
2. **User experience issues** with delayed document processing
3. **Scalability limitations** as transaction volume increases
4. **Lock contention** on inventory tables during concurrent operations
5. **Difficulty tracking historical inventory states** for auditing and analysis

### Business Impact
- Document processing time increases linearly with inventory complexity
- Users experience delays during order placement and shipment processing
- System cannot handle peak transaction loads efficiently
- Limited ability to analyze historical inventory trends
- Difficult to identify root causes of inventory discrepancies

---

## Solution Overview

### Architecture Changes
Implement a **hybrid event-based architecture** with three core components:

1. **M_Reservation Table**: Historical tracking of all inventory commitments and releases
2. **M_Storage Table**: Materialized view of current inventory availability (On Hand, Reserved, Ordered)
3. **Asynchronous Queue Processing**: Decoupled inventory calculations via job queue

### Key Design Principles
- **Event Sourcing**: Every reservation change is recorded as an immutable event
- **Eventual Consistency**: Inventory availability is updated asynchronously after document processing
- **Snapshot Capability**: Point-in-time inventory states for reporting and reconciliation
- **Queue-Based Processing**: Non-blocking inventory updates with guaranteed processing

---

## Technical Implementation

### 1. New Database Tables

#### M_Reservation (Inventory Reservations History)
**Purpose**: Track all inventory commitments and releases as historical events

**Key Columns**:
- `M_Reservation_ID` (PK): Unique identifier
- `ReservationType`: Type of reservation (SO+/SO-, PO+/PO-, Production, etc.)
- `M_Product_ID`: Product being reserved/released
- `M_Warehouse_ID`: Warehouse location
- `M_Locator_ID`: Specific storage location
- `M_AttributeSetInstance_ID`: Product attributes (lot, serial, expiry)
- `Qty`: Quantity (positive = reservation, negative = release)
- Document references: `C_Order_ID`, `C_OrderLine_ID`, `DD_Order_ID`, `M_InOut_ID`, `M_Production_ID`, etc.
- Audit fields: `Created`, `CreatedBy`, `Updated`, `UpdatedBy`

**Reservation Types**:
- `SO` (Reserve Quantity): Sales order reservation - decreases available stock
- `PO` (Order Quantity): Purchase order expected - increases projected stock  
- Production orders, distribution orders, and movement reservations

**Example Flow**:
```sql
-- Sales order created (10 units reserved)
INSERT INTO M_Reservation (ReservationType, M_Product_ID, Qty, C_Order_ID, ...)
VALUES ('SO', 12345, 10, 98765, ...);

-- Shipment processed (10 units released from reservation)
INSERT INTO M_Reservation (ReservationType, M_Product_ID, Qty, M_InOut_ID, ...)
VALUES ('SO', 12345, -10, 55555, ...);

-- Net effect: Reserved quantity returns to 0
```

#### M_StorageSnapshot (Point-in-Time Inventory)
**Purpose**: Capture complete inventory state at specific moments for reporting and reconciliation

**Key Columns**:
- `M_StorageSnapshot_ID` (PK): Unique identifier
- `M_StorageSnapshotRun_ID` (FK): Link to snapshot execution
- `M_Product_ID`: Product
- `M_Locator_ID`: Storage location
- `M_AttributeSetInstance_ID`: Product attributes
- `QtyOnHand`: Physical quantity (from M_Transaction)
- `QtyReserved`: Reserved quantity (from M_Reservation)
- `QtyOrdered`: Ordered quantity (from M_Reservation PO type)
- `DateTrx`: Snapshot timestamp

**Use Cases**:
- Daily inventory snapshots for historical analysis
- Month-end inventory valuation
- Inventory trend analysis
- Discrepancy investigation
- Audit trail compliance

#### M_StorageSnapshotRun (Snapshot Execution Log)
**Purpose**: Track snapshot execution metadata

**Key Columns**:
- `M_StorageSnapshotRun_ID` (PK): Unique identifier
- `DocumentNo`: Execution number
- `DateLastRun`: Execution timestamp
- `ProductProcesses`: Number of products processed
- `TransactionsProcessed`: Number of transactions aggregated

---

### 2. Queue Processing System

#### Queue Type: SMM_Storage_Update
**Configuration**:
- **Queue Type**: `SMM` (Storage Management)
- **Classname**: `org.solop.queue.storage.StorageUpdate`
- **Processing Mode**: Asynchronous batch processing
- **Frequency**: Every 5 minutes (configurable)
- **Batch Size**: 100 records per batch (configurable)

#### Processing Flow
```
Document Completion (Sales Order, Shipment, etc.)
    ↓
Create M_Reservation entries
    ↓
Add document to queue (SMM_Storage_Update)
    ↓
Continue user workflow (non-blocking)
    ↓
[Background Queue Processor]
    ↓
Extract product IDs from queued documents
    ↓
Rebuild M_Storage for affected products
    ↓
Mark queue entry as processed
```

#### Key Components

**StorageUpdate.java** (Queue Processor):
- Extracts product IDs from queued documents
- Delegates to `StorageUpdaterBuilder` for inventory recalculation
- Supports multiple document types:
  - Sales Orders (`C_Order`)
  - Purchase Orders (via `C_Order`)
  - Distribution Orders (`DD_Order`)
  - Shipments/Receipts (`M_InOut`)
  - Inventory Movements (`M_Movement`)
  - Production Orders (`M_ProductionBatch`)
  - Production Execution (`M_Production`)
  - Project Issues (`C_ProjectIssue`)
  - Physical Inventory (`M_Inventory`)

**StorageUpdaterBuilder.java** (Calculation Engine):
- Deletes existing M_Storage records for affected products
- Recalculates from source data:
  - `QtyOnHand`: SUM(M_Transaction.MovementQty)
  - `QtyReserved`: SUM(M_Reservation.Qty WHERE ReservationType NOT IN ('PO+', 'PO-'))
  - `QtyOrdered`: SUM(M_Reservation.Qty WHERE ReservationType IN ('PO+', 'PO-'))
- Groups by Product, Locator, AttributeSetInstance
- Creates new M_Storage records with calculated values

---

### 3. Document Integration

Each transactional document now follows this pattern:

#### Sales Order (MOrder.java / MOrderLine.java)
```java
@Override
protected boolean afterSave(boolean newRecord, boolean success) {
    // Create reservation when order is saved
    ReservationBuilder.newInstance(getCtx(), get_TrxName())
        .withSalesOrderLine(this)
        .build();
    return success;
}
```

**Reservation Events**:
- Order created → Create positive reservation (`ReservationType = 'SO'`)
- Order modified → Adjust reservation (delta calculation)
- Order voided → Create negative reservation to release
- Shipment created → Create negative reservation to release

#### Shipment/Receipt (MInOut.java)
```java
@Override
protected boolean afterComplete(String DocAction) {
    // Add to storage update queue
    StorageUpdate.addDocumentToQueue(this);
    return true;
}
```

**Processing**:
1. Shipment completed → Creates M_Transaction records
2. Releases reservations → Negative M_Reservation entries
3. Adds to queue → Async M_Storage update

#### Production Orders (MProductionBatch.java / MProductionBatchLine.java)
```java
@Override
protected boolean afterSave(boolean newRecord, boolean success) {
    // Reserve raw materials (negative for components)
    // Reserve production output (positive for finished goods)
    ReservationBuilder.newInstance(getCtx(), get_TrxName())
        .withProductionOrderLine(this)
        .build();
    return success;
}
```

**Material Flow**:
- Production order created → Reserve raw materials (SO type)
- Production order created → Reserve output location (PO type for finished goods)
- Production executed → Release reservations, create transactions

#### Distribution Orders (MOrder with DD_Order type)
**Reservations**:
- From warehouse: `ReservationType = 'SO'` (reserve at source)
- To warehouse: `ReservationType = 'PO'` (expected at destination)

---

### 4. Snapshot Process (UpdateStorage.java)

#### Purpose
Create point-in-time inventory snapshots and rebuild complete storage from historical data.

#### Execution Modes

**Mode 1: Selective Update** (Daily Operations)
```java
// Update specific products/warehouses
StorageUpdaterBuilder.newInstance(ctx, trxName)
    .withWarehouseId(warehouseId)
    .withProductId(productId)
    .build();
```

**Mode 2: Full Snapshot** (Periodic Reconciliation)
```java
// Create complete system snapshot
UpdateStorage process = new UpdateStorage();
process.setIsCreateSnapshot(true);
process.doIt();
```

#### Snapshot Creation Algorithm

**Step 1: Snapshot from Reservations**
```sql
INSERT INTO M_StorageSnapshot (
    M_Product_ID, M_Locator_ID, M_AttributeSetInstance_ID,
    QtyOnHand, QtyReserved, QtyOrdered, M_StorageSnapshotRun_ID, DateTrx
)
SELECT 
    r.M_Product_ID,
    r.M_Locator_ID,
    r.M_AttributeSetInstance_ID,
    0 as QtyOnHand,
    SUM(CASE WHEN r.ReservationType NOT IN('PO+', 'PO-') THEN r.Qty ELSE 0 END) as QtyReserved,
    SUM(CASE WHEN r.ReservationType IN('PO+', 'PO-') THEN r.Qty ELSE 0 END) as QtyOrdered,
    ?, -- M_StorageSnapshotRun_ID
    ?  -- Current timestamp
FROM M_Reservation r
GROUP BY r.M_Product_ID, r.M_Locator_ID, r.M_AttributeSetInstance_ID
HAVING SUM(r.Qty) <> 0;
```

**Step 2: Snapshot from Transactions**
```sql
INSERT INTO M_StorageSnapshot (
    M_Product_ID, M_Locator_ID, M_AttributeSetInstance_ID,
    QtyOnHand, QtyReserved, QtyOrdered, M_StorageSnapshotRun_ID, DateTrx
)
SELECT 
    mt.M_Product_ID,
    mt.M_Locator_ID,
    mt.M_AttributeSetInstance_ID,
    SUM(mt.MovementQty) as QtyOnHand,
    0 as QtyReserved,
    0 as QtyOrdered,
    ?, -- M_StorageSnapshotRun_ID
    ?  -- Current timestamp
FROM M_Transaction mt
INNER JOIN M_Locator ml ON mt.M_Locator_ID = ml.M_Locator_ID
GROUP BY mt.M_Product_ID, mt.M_Locator_ID, mt.M_AttributeSetInstance_ID
HAVING SUM(mt.MovementQty) <> 0;
```

**Step 3: Rebuild M_Storage**
```java
StorageUpdaterBuilder.newInstance(ctx, trxName)
    .withClientId(clientId)
    .build(); // Rebuilds all storage records
```

#### Scheduled Jobs

**Job 1: Queue Processor** (FlushSystemQueue)
- **Frequency**: Every 5 minutes
- **Purpose**: Process pending storage updates
- **Parameters**:
  - Batches to process: 10
  - Records per batch: 100
  - Delete after process: No
  - Queue Type: SMM_Storage_Update

**Job 2: Snapshot Creator** (UpdateStorage)
- **Frequency**: Daily (configurable)
- **Purpose**: Create full inventory snapshot and rebuild storage
- **Parameters**:
  - Create Snapshot: Yes

---

## Code Changes Summary

### New Classes
1. **MReservation.java** - Reservation entity model
2. **MStorageSnapshot.java** - Snapshot entity model
3. **MStorageSnapshotRun.java** - Snapshot run entity model
4. **StorageUpdate.java** - Queue processor for storage updates
5. **UpdateStorage.java** - Snapshot creation and storage rebuild process
6. **StorageUpdaterBuilder.java** - Storage calculation builder
7. **CreateStorageUpdateSchedule.java** - Setup utility for scheduled jobs

### Modified Classes
1. **MOrder.java / MOrderLine.java** - Added reservation creation
2. **MInOut.java** - Added queue integration
3. **MMovement.java** - Added queue integration
4. **MProduction.java** - Added queue integration
5. **MProductionBatch.java / MProductionBatchLine.java** - Added reservation logic
6. **MInventory.java** - Added queue integration
7. **MProjectIssue.java** - Added queue integration
8. **MStorage.java** - Modified to work with async updates

### New Database Tables
1. `M_Reservation` - Reservation history
2. `M_StorageSnapshot` - Point-in-time snapshots
3. `M_StorageSnapshotRun` - Snapshot execution log

### New Queue Type
- **SMM_Storage_Update** (Storage Management Module)

---

## Benefits

### Performance Improvements
1. **Faster Document Processing**: Document completion no longer blocks on inventory calculations
2. **Reduced Lock Contention**: Inventory updates happen in background batches
3. **Scalability**: System can handle 10x more concurrent transactions
4. **Predictable Response Times**: User operations complete in milliseconds instead of seconds

### Operational Benefits
1. **Historical Tracking**: Complete audit trail of all reservation events
2. **Snapshot Capability**: Point-in-time inventory for reporting and reconciliation
3. **Easier Debugging**: Can trace exact sequence of events affecting inventory
4. **Flexible Reporting**: Query historical inventory states for any date/time

### Data Quality
1. **Eventually Consistent**: Storage always reflects true state after queue processing
2. **Reconciliation**: Snapshots enable automated discrepancy detection
3. **Audit Trail**: Every reservation change is logged with user and timestamp

---

## Migration Strategy

### Phase 1: Initial Deployment
1. Deploy new database tables (M_Reservation, M_StorageSnapshot, M_StorageSnapshotRun)
2. Deploy queue infrastructure (SMM_Storage_Update queue type)
3. Deploy code changes to document models
4. Create scheduled jobs (queue processor, snapshot creator)

### Phase 2: Historical Data Migration
1. Run initial snapshot to baseline current inventory
2. Validate snapshot against current M_Storage
3. Archive old M_Storage records (if needed)

### Phase 3: Monitoring & Optimization
1. Monitor queue processing times
2. Adjust batch sizes based on load
3. Tune snapshot frequency
4. Monitor storage calculation accuracy

### Rollback Plan
If issues arise:
1. Disable queue processor scheduled job
2. Revert to synchronous storage updates (legacy code path)
3. Investigate and fix issues
4. Re-enable asynchronous processing

---

## Testing Requirements

### Unit Tests
- [ ] Reservation creation for each document type
- [ ] Reservation reversal on document void/delete
- [ ] Queue entry creation
- [ ] Storage calculation from reservations
- [ ] Storage calculation from transactions
- [ ] Snapshot creation

### Integration Tests
- [ ] End-to-end sales order → shipment flow
- [ ] End-to-end purchase order → receipt flow
- [ ] Production order → production execution
- [ ] Distribution order → movement
- [ ] Concurrent order processing (load test)
- [ ] Queue processing under high load

### Reconciliation Tests
- [ ] Compare M_Storage vs snapshot
- [ ] Compare M_Storage vs raw calculations
- [ ] Verify reservation totals match expectations
- [ ] Historical snapshot accuracy

---

## Configuration

### System Properties
```properties
# Queue Processing
queue.storage.batch.size=100
queue.storage.frequency.minutes=5
queue.storage.delete.after.process=false

# Snapshot
snapshot.frequency.days=1
snapshot.retention.days=90
```

### Scheduled Jobs Setup
```java
// Auto-created by CreateStorageUpdateSchedule.java
// 1. Queue Processor: Every 5 minutes
// 2. Snapshot Creator: Daily at midnight
```

---

## Monitoring & Metrics

### Key Metrics to Track
1. **Queue Processing Time**: Average time per batch
2. **Queue Depth**: Number of pending documents
3. **Storage Calculation Time**: Time to rebuild M_Storage per product
4. **Snapshot Duration**: Time to create full system snapshot
5. **Discrepancies**: Differences between snapshots and expected values

### Dashboard Queries
```sql
-- Queue backlog
SELECT COUNT(*) FROM AD_Queue WHERE AD_QueueType_ID = 50016 AND Processed = 'N';

-- Average processing time
SELECT AVG(Updated - Created) FROM AD_Queue WHERE AD_QueueType_ID = 50016 AND Processed = 'Y';

-- Latest snapshot stats
SELECT * FROM M_StorageSnapshotRun ORDER BY DateLastRun DESC LIMIT 1;

-- Products with discrepancies
SELECT s.M_Product_ID, 
       s.QtyOnHand as StorageQty,
       ss.QtyOnHand as SnapshotQty,
       s.QtyOnHand - ss.QtyOnHand as Difference
FROM M_Storage s
LEFT JOIN M_StorageSnapshot ss ON s.M_Product_ID = ss.M_Product_ID 
                                AND s.M_Locator_ID = ss.M_Locator_ID
WHERE ABS(s.QtyOnHand - ss.QtyOnHand) > 0.001;
```

---

## Known Limitations

1. **Eventual Consistency**: Brief lag between document completion and inventory availability update
2. **Queue Dependency**: System depends on queue processor running reliably
3. **Storage During Migration**: Cannot run old and new system simultaneously

---

## Future Enhancements

1. **Real-time Availability**: Add Redis cache for instant availability checks
2. **Event Streaming**: Integrate with Kafka for event distribution
3. **Predictive Analytics**: Use reservation history for demand forecasting
4. **Automated Reconciliation**: Auto-detect and alert on inventory discrepancies
5. **Distributed Processing**: Scale queue processors horizontally

---

## Documentation Updates Required

1. User Manual: Explain asynchronous processing behavior
2. Technical Documentation: Architecture diagrams and data flow
3. API Documentation: New reservation and snapshot endpoints
4. Migration Guide: Steps for existing installations
5. Troubleshooting Guide: Common issues and resolutions

---

## References

### Related Issues
- Performance improvements for high-volume transactions
- Inventory audit trail requirements
- Historical inventory reporting

### Design Documents
- Event Sourcing Architecture Pattern
- Queue-Based Processing Best Practices
- Inventory Management Data Model

### External Resources
- ADempiere Wiki: Storage Management
- Queue Framework Documentation
- Database Performance Tuning Guide

---

## Implementation Checklist

- [x] Database schema design
- [x] M_Reservation table implementation
- [x] M_StorageSnapshot table implementation
- [x] Queue processor implementation
- [x] Storage update builder implementation
- [x] Document model integration (Orders, Shipments, Production, etc.)
- [x] Snapshot process implementation
- [x] Scheduled job setup automation
- [ ] Unit tests
- [ ] Integration tests
- [ ] Performance tests
- [ ] Documentation updates
- [ ] Migration scripts
- [ ] Deployment guide

---

## Contributors
- **Developer**: [Your Name]
- **Reviewers**: [Team Members]
- **Testing**: [QA Team]

## Version History
- **v1.0** - Initial implementation
- **Date**: January 22, 2026

---

## Approval Sign-off

- [ ] Technical Lead
- [ ] Architecture Review
- [ ] QA Sign-off
- [ ] Product Owner Approval

Uh oh!

[Feature Request] Storage as Queue #533

Description

Asynchronous Inventory Processing Implementation

Issue Type

Priority

Summary

Background

Problem Statement

Business Impact

Solution Overview

Architecture Changes

Key Design Principles

Technical Implementation

1. New Database Tables

M_Reservation (Inventory Reservations History)

M_StorageSnapshot (Point-in-Time Inventory)

M_StorageSnapshotRun (Snapshot Execution Log)

2. Queue Processing System

Queue Type: SMM_Storage_Update

Processing Flow

Key Components

3. Document Integration

Sales Order (MOrder.java / MOrderLine.java)

Shipment/Receipt (MInOut.java)

Production Orders (MProductionBatch.java / MProductionBatchLine.java)

Distribution Orders (MOrder with DD_Order type)

4. Snapshot Process (UpdateStorage.java)

Purpose

Execution Modes

Snapshot Creation Algorithm

Scheduled Jobs

Code Changes Summary

New Classes

Modified Classes

New Database Tables

New Queue Type

Benefits

Performance Improvements

Operational Benefits

Data Quality

Migration Strategy

Phase 1: Initial Deployment

Phase 2: Historical Data Migration

Phase 3: Monitoring & Optimization

Rollback Plan

Testing Requirements

Unit Tests

Integration Tests

Reconciliation Tests

Configuration

System Properties

Scheduled Jobs Setup

Monitoring & Metrics

Key Metrics to Track

Dashboard Queries

Known Limitations

Future Enhancements

Documentation Updates Required

References

Related Issues

Design Documents

External Resources

Implementation Checklist

Contributors

Version History

Approval Sign-off

Metadata

Metadata

Assignees

Labels

Type

Fields

Projects

Milestone

Relationships

Development

Issue actions