Stock-Keeper Microservice

High-throughput inventory ledger built with Framework M + TigerBeetle.

Duration: 3 days Owner: Dev 1 (Framework track)

---

Overview

Stock-Keeper is a dedicated microservice for inventory transactions:

• Receives stock movements from WMS
• Maintains real-time balances via TigerBeetle
• Provides query APIs for stock levels, ledger, projections

┌─────────────────────────────────────────────────────────────┐
│                      Legacy ERP WMS App                             │
│                                                              │
│  StockEntry ──► m-ledger adapter ──► Stock-Keeper Client    │
└──────────────────────────────────────────────────────────────┘
                                              │
                                         gRPC │
                                              ▼
┌────────────────────────────────────────────────────────────┐
│                     Stock-Keeper                           │
│  ┌──────────────────────────────────────────────────────┐  │
│  │                   Litestar + gRPC                    │  │
│  ├──────────────────────────────────────────────────────┤  │
│  │  POST /stock/transfer    - Move stock                │  │
│  │  GET  /stock/balance     - Current balance           │  │
│  │  GET  /stock/ledger      - Transaction history       │  │
│  │  GET  /stock/projected   - Balance + pending         │  │
│  └──────────────────────────────────────────────────────┘  │
│                            │                               │
│                     ┌──────▼──────┐                        │
│                     │ TigerBeetle │                        │
│                     │   Cluster   │                        │
│                     └─────────────┘                        │
└────────────────────────────────────────────────────────────┘

---

1. Project Structure

stock-keeper/
├── pyproject.toml
├── m.toml                       # Framework M config
├── stock_keeper/
│   ├── __init__.py
│   ├── app.py                   # Litestar/gRPC entrypoint
│   ├── config.py                # Settings
│   ├── models/
│   │   ├── account.py           # Warehouse/Bin as TB account
│   │   ├── transfer.py          # Stock movement as TB transfer
│   │   └── schemas.py           # Pydantic schemas
│   ├── services/
│   │   ├── stock_service.py     # Core business logic
│   │   └── tigerbeetle.py       # TB client wrapper
│   ├── api/
│   │   ├── rest.py              # REST endpoints
│   │   └── grpc.py              # gRPC service
│   └── proto/
│       └── stock.proto          # gRPC definitions
├── tests/
│   ├── test_stock_service.py
│   └── test_api.py
└── Dockerfile

---

2. m.toml Configuration

[app]
name = "stock-keeper"
version = "0.1.0"
description = "High-throughput inventory ledger"

[database]
# No Postgres — TigerBeetle only
backend = "tigerbeetle"
cluster_id = 1
addresses = ["127.0.0.1:3001"]

[server]
host = "0.0.0.0"
port = 8001
workers = 4

[server.grpc]
enabled = true
port = 50051

[features]
doctypes = false      # No ORM-based DocTypes
admin_ui = false      # No Refine frontend
migrations = false    # TB handles schema

---

3. TigerBeetle Account Model

Map inventory concepts to TigerBeetle:

Inventory Concept	TigerBeetle	Account Code Format
Warehouse	Account	{tenant_id}-WH-{warehouse}
Bin (location)	Account	{tenant_id}-BIN-{warehouse}-{bin}
Item	Ledger	Separate ledger per item
In-Transit	Account	{tenant_id}-TRANSIT-{from}-{to}
Adjustment	Account	{tenant_id}-ADJ

from tigerbeetle import Account, Transfer
from uuid import uuid5, NAMESPACE_DNS

def warehouse_account_id(tenant_id: str, warehouse: str) -> int:
    """Generate deterministic 128-bit account ID."""
    name = f"{tenant_id}-WH-{warehouse}"
    return int(uuid5(NAMESPACE_DNS, name).int)

def create_warehouse_account(tenant_id: str, warehouse: str) -> Account:
    return Account(
        id=warehouse_account_id(tenant_id, warehouse),
        ledger=1,  # Stock ledger
        code=0,
        flags=0,
    )

---

4. Stock Service

from decimal import Decimal
from tigerbeetle import Client, Transfer, TransferFlags

class StockService:
    """Core stock operations backed by TigerBeetle."""

    def __init__(self, tb_client: Client):
        self.tb = tb_client

    async def transfer_stock(
        self,
        tenant_id: str,
        item: str,
        from_warehouse: str,
        to_warehouse: str,
        qty: Decimal,
        rate: Decimal,
        reference: str,
    ) -> str:
        """Move stock between warehouses."""
        transfer = Transfer(
            id=generate_transfer_id(),
            debit_account_id=warehouse_account_id(tenant_id, from_warehouse),
            credit_account_id=warehouse_account_id(tenant_id, to_warehouse),
            amount=int(qty * 1000),  # 3 decimal precision
            ledger=item_ledger_id(item),
            code=TransferCode.STOCK_TRANSFER,
            user_data_128=encode_reference(reference),
        )
        result = await self.tb.create_transfers([transfer])
        if result:
            raise StockError(f"Transfer failed: {result[0]}")
        return str(transfer.id)

    async def get_balance(
        self,
        tenant_id: str,
        item: str,
        warehouse: str,
    ) -> Decimal:
        """Get current stock balance."""
        account_id = warehouse_account_id(tenant_id, warehouse)
        accounts = await self.tb.lookup_accounts([account_id])
        if not accounts:
            return Decimal("0")
        # Credits = receipts, Debits = issues
        balance = accounts[0].credits_posted - accounts[0].debits_posted
        return Decimal(balance) / 1000

    async def get_ledger(
        self,
        tenant_id: str,
        item: str,
        warehouse: str,
        limit: int = 100,
    ) -> list[dict]:
        """Get stock transaction history."""
        account_id = warehouse_account_id(tenant_id, warehouse)
        transfers = await self.tb.get_account_transfers(
            account_id,
            limit=limit,
        )
        return [format_transfer(t) for t in transfers]

---

5. REST API

from litestar import Litestar, get, post
from litestar.exceptions import HTTPException
from msgspec import Struct

# Using msgspec for high-performance serialization

class StockTransferRequest(Struct):
    tenant_id: str
    item: str
    from_warehouse: str
    to_warehouse: str
    qty: Decimal
    rate: Decimal
    reference: str

class StockBalanceResponse(Struct):
    item: str
    warehouse: str
    qty: Decimal
    as_of: datetime

@post("/stock/transfer")
async def transfer_stock(req: StockTransferRequest) -> dict:
    transfer_id = await stock_service.transfer_stock(
        tenant_id=req.tenant_id,
        item=req.item,
        from_warehouse=req.from_warehouse,
        to_warehouse=req.to_warehouse,
        qty=req.qty,
        rate=req.rate,
        reference=req.reference,
    )
    return {"transfer_id": transfer_id}

@get("/stock/balance/{tenant_id:str}/{item:str}/{warehouse:str}")
async def get_balance(tenant_id: str, item: str, warehouse: str) -> StockBalanceResponse:
    qty = await stock_service.get_balance(tenant_id, item, warehouse)
    return StockBalanceResponse(
        item=item,
        warehouse=warehouse,
        qty=qty,
        as_of=datetime.utcnow(),
    )

@get("/stock/ledger/{tenant_id:str}/{item:str}/{warehouse:str}")
async def get_ledger(tenant_id: str, item: str, warehouse: str, limit: int = 100) -> list[dict]:
    return await stock_service.get_ledger(tenant_id, item, warehouse, limit)

# Create Litestar app
app = Litestar(route_handlers=[transfer_stock, get_balance, get_ledger])

---

6. gRPC Service

proto/stock.proto

syntax = "proto3";
package stockkeeper;

service StockService {
    rpc Transfer(TransferRequest) returns (TransferResponse);
    rpc GetBalance(BalanceRequest) returns (BalanceResponse);
    rpc GetLedger(LedgerRequest) returns (LedgerResponse);
    rpc BatchTransfer(BatchTransferRequest) returns (BatchTransferResponse);
}

message TransferRequest {
    string tenant_id = 1;
    string item = 2;
    string from_warehouse = 3;
    string to_warehouse = 4;
    string qty = 5;  // Decimal as string
    string rate = 6;
    string reference = 7;
}

message TransferResponse {
    string transfer_id = 1;
}

message BalanceRequest {
    string tenant_id = 1;
    string item = 2;
    string warehouse = 3;
}

message BalanceResponse {
    string item = 1;
    string warehouse = 2;
    string qty = 3;
    string as_of = 4;
}

---

7. m-ledger Integration

The WMS app uses Stock-Keeper via the m-ledger adapter:

modules/m_ledger/adapters/stock_keeper.py

class StockKeeperAdapter(LedgerProtocol):
    """Stock-Keeper client for m-ledger."""

    def __init__(self, base_url: str = "http://stock-keeper:8001"):
        self.client = httpx.AsyncClient(base_url=base_url)

    async def post_entry(
        self,
        tenant_id: str,
        debit_account: str,  # from_warehouse
        credit_account: str,  # to_warehouse
        amount: Decimal,
        **kwargs,
    ) -> str:
        response = await self.client.post("/stock/transfer", json={
            "tenant_id": tenant_id,
            "item": kwargs["item"],
            "from_warehouse": debit_account,
            "to_warehouse": credit_account,
            "qty": str(amount),
            "rate": str(kwargs.get("rate", 0)),
            "reference": kwargs.get("voucher_no", ""),
        })
        return response.json()["transfer_id"]

    async def get_balance(self, tenant_id: str, account: str, **kwargs) -> Decimal:
        item = kwargs["item"]
        response = await self.client.get(f"/stock/balance/{tenant_id}/{item}/{account}")
        return Decimal(response.json()["qty"])

---

8. Deployment

docker-compose.yml

services:
  stock-keeper:
    build: ./stock-keeper
    command: m start
    ports:
      - "8001:8001"
      - "50051:50051"
    environment:
      - TIGERBEETLE_ADDRESSES=tigerbeetle:3001
    depends_on:
      - tigerbeetle

  tigerbeetle:
    image: tigerbeetle/tigerbeetle:latest
    command: start --addresses=0.0.0.0:3001 /data/0_0.tigerbeetle
    volumes:
      - tb_stock_data:/data
    ports:
      - "3001:3001"

volumes:
  tb_stock_data:

---

9. Performance Targets

Metric	Target
Throughput	10,000 TPS (TigerBeetle max: 1M TPS)
Latency (p99)	< 10ms
Balance query	< 1ms
Ledger query (100 items)	< 5ms

---

10. Task Checklist

• Project setup (m.toml, pyproject.toml)
• TigerBeetle client wrapper
• Account ID generation strategy
• StockService core operations
• REST API endpoints
• gRPC service (optional, for high-perf)
• m-ledger StockKeeperAdapter
• Docker + compose setup
• Load tests (k6)
• Integration with WMS

---

11. CLI

# Start Stock-Keeper
cd stock-keeper
m start

# Or with Docker
docker-compose up stock-keeper

# Health check
curl http://localhost:8001/health

# Test transfer
curl -X POST http://localhost:8001/stock/transfer \
  -H "Content-Type: application/json" \
  -d '{"item": "ITEM-001", "from_warehouse": "WH-DEL-MAIN", "to_warehouse": "WH-DEL-BIN1",
  "qty": "100",
  "rate": "50.00",
  "reference": "GRN-001"}'