feat: Enhance audit trail with created_by and updated_by fields across services

- Added `created_by_username` and `updated_by_username` fields to relevant schemas and models for tracking user actions.
- Implemented utility functions to extract and format audit fields for consistency across services.
- Updated service methods to include audit information during create and update operations.
- Refactored customer and staff controllers to pass user information for audit purposes.
- Introduced `MetaSchema` for standardized audit information in API responses.
- Enhanced documentation for new schemas and utility functions.

POS_META_SCHEMA_IMPLEMENTATION.md

@@ -0,0 +1,563 @@
+# POS Microservice: Meta Schema Implementation
+
+## Overview
+
+This document describes the implementation of the **Meta Schema Pattern** in the POS microservice. This pattern groups audit fields (created_by, created_at, updated_by, updated_at) under a single 'meta' object in API responses, providing consistent audit trail information across all entities.
+
+## Implementation Date
+
+December 2024
+
+## Modules Updated
+
+1. **Customers Module** (`app/customers/`)
+2. **Staff Module** (`app/staff/`)
+
+## Core Infrastructure
+
+### 1. Core Schemas (`app/core/schemas.py`)
+
+Added the following reusable schemas:
+
+- **MetaSchema**: Defines the structure for audit information
+- **PaginationMeta**: Pagination information for list endpoints
+- **ErrorDetail**: Standard error detail structure
+- **SuccessResponse**: Standard success response
+- **StatusResponse**: Standard status response with optional data payload
+- **BaseLazyFetchSchema**: Base schema for list requests with pagination
+
+### 2. Core Utilities (`app/core/utils.py`)
+
+Three essential utility functions:
+
+#### format_meta_field(document: dict) -> dict
+Transforms database documents to API response format by:
+- Extracting audit fields from document
+- Creating 'meta' object with proper field mapping
+- Removing audit fields from top level
+- Returning formatted document
+
+**Field Mapping:**
+```
+Database Field          → Response Field
+----------------          ----------------
+created_by (UUID)       → meta.created_by_id
+created_by_username     → meta.created_by
+created_at              → meta.created_at
+updated_by (UUID)       → meta.updated_by_id
+updated_by_username     → meta.updated_by
+updated_at              → meta.updated_at
+```
+
+#### extract_audit_fields(user_id: str, username: str, is_update: bool) -> dict
+Creates audit field dictionaries for database operations:
+- For creation (`is_update=False`): Returns created_by, created_by_username, created_at
+- For updates (`is_update=True`): Returns updated_by, updated_by_username, updated_at
+
+#### normalize_uuid_fields(document: dict, fields: List[str]) -> dict
+Converts UUID objects to strings for JSON serialization.
+
+### 3. Core Documentation (`app/core/README.md`)
+
+Comprehensive documentation with:
+- Usage examples for all entities
+- Implementation guide for new modules
+- Benefits and best practices
+- Migration checklist
+
+## Customers Module Implementation
+
+### Changes Made
+
+#### 1. Model (`app/customers/models/model.py`)
+
+**Added Audit Fields:**
+```python
+class CustomerModel(BaseModel):
+    # ... existing fields ...
+    
+    # Audit fields
+    created_by: str = Field(..., description="User ID who created")
+    created_by_username: Optional[str] = Field(None, description="Username who created")
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_by: Optional[str] = Field(None, description="User ID who updated")
+    updated_by_username: Optional[str] = Field(None, description="Username who updated")
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+```
+
+#### 2. Schemas (`app/customers/schemas/schema.py`)
+
+**Updated CustomerCreate:**
+```python
+class CustomerCreate(CustomerBase):
+    merchant_id: Optional[str] = Field(None, description="Merchant identifier (UUID) - will be overridden by token")
+    created_by_username: Optional[str] = Field(None, description="Username who created - will be set from token")
+```
+
+**Updated CustomerUpdate:**
+```python
+class CustomerUpdate(BaseModel):
+    # ... existing fields ...
+    updated_by_username: Optional[str] = Field(None, description="Username who updated - will be set from token")
+```
+
+**Updated CustomerResponse:**
+```python
+class CustomerResponse(CustomerBase):
+    customer_id: str = Field(..., description="Customer identifier (UUID)")
+    meta: Dict[str, Any] = Field(..., description="Audit information (created_by, created_at, updated_by, updated_at)")
+    
+    class Config:
+        from_attributes = True
+```
+
+#### 3. Service (`app/customers/services/service.py`)
+
+**Added Imports:**
+```python
+from app.core.utils import format_meta_field, extract_audit_fields
+```
+
+**Updated create_customer:**
+- Added `user_id` and `username` parameters
+- Uses `extract_audit_fields(is_update=False)` to add audit fields
+- Uses `format_meta_field()` before returning CustomerResponse
+
+**Updated get_customer:**
+- Uses `format_meta_field()` before returning CustomerResponse
+
+**Updated list_customers:**
+- Uses `format_meta_field()` for each document when not using projection
+- Returns raw dicts when projection is used (for performance)
+
+**Updated update_customer:**
+- Added `user_id` and `username` parameters
+- Uses `extract_audit_fields(is_update=True)` to add update audit fields
+- Uses `format_meta_field()` before returning CustomerResponse
+
+**Updated delete_customer:**
+- Added `user_id` and `username` parameters
+- Uses `extract_audit_fields(is_update=True)` for soft delete audit trail
+- Uses `format_meta_field()` before returning CustomerResponse
+
+#### 4. Controller (`app/customers/controllers/router.py`)
+
+**Updated create_customer endpoint:**
+```python
+async def create_customer(
+    payload: CustomerCreate,
+    current_user: TokenUser = Depends(require_pos_permission("customers", "create"))
+):
+    payload.created_by_username = current_user.username
+    customer = await CustomerService.create_customer(
+        payload, 
+        current_user.merchant_id,
+        current_user.user_id,
+        current_user.username
+    )
+```
+
+**Updated update_customer endpoint:**
+```python
+async def update_customer(
+    customer_id: str, 
+    payload: CustomerUpdate,
+    current_user: TokenUser = Depends(require_pos_permission("customers", "update"))
+):
+    payload.updated_by_username = current_user.username
+    customer = await CustomerService.update_customer(
+        customer_id=customer_id,
+        payload=payload,
+        merchant_id=current_user.merchant_id,
+        merchant_type=current_user.merchant_type,
+        user_id=current_user.user_id,
+        username=current_user.username
+    )
+```
+
+**Updated delete_customer endpoint:**
+```python
+async def delete_customer(
+    customer_id: str,
+    current_user: TokenUser = Depends(require_pos_permission("customers", "delete"))
+):
+    await CustomerService.delete_customer(
+        customer_id=customer_id,
+        merchant_id=current_user.merchant_id,
+        merchant_type=current_user.merchant_type,
+        user_id=current_user.user_id,
+        username=current_user.username
+    )
+```
+
+### API Response Example (Customers)
+
+**Before (Old Format):**
+```json
+{
+  "customer_id": "cus_123",
+  "name": "John Doe",
+  "phone": "+91-9988776655",
+  "email": "john@example.com",
+  "status": "active",
+  "created_at": "2023-01-10T08:00:00Z",
+  "updated_at": "2024-11-24T10:00:00Z"
+}
+```
+
+**After (Meta Format):**
+```json
+{
+  "customer_id": "cus_123",
+  "name": "John Doe",
+  "phone": "+91-9988776655",
+  "email": "john@example.com",
+  "status": "active",
+  "meta": {
+    "created_by": "admin",
+    "created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
+    "created_at": "2023-01-10T08:00:00Z",
+    "updated_by": "manager",
+    "updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
+    "updated_at": "2024-11-24T10:00:00Z"
+  }
+}
+```
+
+## Staff Module Implementation
+
+### Changes Made
+
+#### 1. Model (`app/staff/models/staff_model.py`)
+
+**Added Audit Fields:**
+```python
+class StaffModel(BaseModel):
+    # ... existing fields ...
+    
+    # Audit fields
+    created_by: str = Field(..., description="User ID who created")
+    created_by_username: Optional[str] = Field(None, description="Username who created")
+    created_at: datetime = Field(default_factory=datetime.utcnow, description="Creation timestamp")
+    updated_by: Optional[str] = Field(None, description="User ID who updated")
+    updated_by_username: Optional[str] = Field(None, description="Username who updated")
+    updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update timestamp")
+```
+
+#### 2. Schemas (`app/staff/schemas/staff_schema.py`)
+
+**Updated StaffCreateSchema:**
+```python
+class StaffCreateSchema(BaseModel):
+    # ... existing fields ...
+    created_by_username: Optional[str] = Field(None, description="Username who created - will be set from token")
+```
+
+**Updated StaffUpdateSchema:**
+```python
+class StaffUpdateSchema(BaseModel):
+    # ... existing fields ...
+    updated_by_username: Optional[str] = Field(None, description="Username who updated - will be set from token")
+```
+
+**Updated StaffResponseSchema:**
+```python
+class StaffResponseSchema(BaseModel):
+    staff_id: UUID
+    merchant_id: UUID
+    name: str
+    role: str
+    # ... other fields ...
+    meta: dict = Field(..., description="Audit information (created_by, created_at, updated_by, updated_at)")
+    
+    class Config:
+        from_attributes = True
+```
+
+#### 3. Service (`app/staff/services/staff_service.py`)
+
+**Added Imports:**
+```python
+from app.core.utils import format_meta_field, extract_audit_fields
+```
+
+**Updated create_staff:**
+- Added `user_id` and `username` parameters
+- Uses `extract_audit_fields(is_update=False)` to add audit fields
+- Uses `format_meta_field()` before returning StaffResponseSchema
+
+**Updated get_staff_by_id:**
+- Uses `format_meta_field()` before returning StaffResponseSchema
+
+**Updated update_staff:**
+- Added `user_id` and `username` parameters
+- Uses `extract_audit_fields(is_update=True)` to add update audit fields
+- Uses `format_meta_field()` before returning StaffResponseSchema (via get_staff_by_id)
+
+**Note:** list_staff returns raw dicts and doesn't need formatting (projection-based approach).
+
+#### 4. Controller (`app/staff/controllers/router.py`)
+
+**Updated create_staff endpoint:**
+```python
+async def create_staff(
+    payload: StaffCreateSchema,
+    current_user: TokenUser = Depends(require_pos_permission("staff", "create"))
+):
+    payload.created_by_username = current_user.username
+    result = await StaffService.create_staff(
+        payload,
+        user_id=current_user.user_id,
+        username=current_user.username
+    )
+```
+
+**Updated update_staff endpoint:**
+```python
+async def update_staff(
+    staff_id: UUID,
+    payload: StaffUpdateSchema,
+    current_user: TokenUser = Depends(require_pos_permission("staff", "edit"))
+):
+    payload.updated_by_username = current_user.username
+    result = await StaffService.update_staff(
+        staff_id, 
+        current_user.merchant_id, 
+        payload,
+        user_id=current_user.user_id,
+        username=current_user.username
+    )
+```
+
+**Updated update_staff_status endpoint:**
+```python
+async def update_staff_status(
+    staff_id: UUID,
+    payload: UpdateStatusRequest,
+    current_user: TokenUser = Depends(require_pos_permission("staff", "edit"))
+):
+    update_payload = StaffUpdateSchema(
+        status=payload.status,
+        updated_by_username=current_user.username
+    )
+    result = await StaffService.update_staff(
+        staff_id, 
+        current_user.merchant_id, 
+        update_payload,
+        user_id=current_user.user_id,
+        username=current_user.username
+    )
+```
+
+### API Response Example (Staff)
+
+**Before (Old Format):**
+```json
+{
+  "staff_id": "staff_123",
+  "merchant_id": "merchant_789",
+  "name": "Aarav Sharma",
+  "role": "Stylist",
+  "phone": "+91-9876543210",
+  "email": "aarav@retail.com",
+  "status": "active",
+  "created_at": "2024-01-15T10:30:00Z",
+  "updated_at": "2024-01-15T10:30:00Z"
+}
+```
+
+**After (Meta Format):**
+```json
+{
+  "staff_id": "staff_123",
+  "merchant_id": "merchant_789",
+  "name": "Aarav Sharma",
+  "role": "Stylist",
+  "phone": "+91-9876543210",
+  "email": "aarav@retail.com",
+  "status": "active",
+  "meta": {
+    "created_by": "admin",
+    "created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
+    "created_at": "2024-01-15T10:30:00Z",
+    "updated_by": "manager",
+    "updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
+    "updated_at": "2024-12-01T14:20:00Z"
+  }
+}
+```
+
+## Benefits
+
+### 1. Consistency
+All entities (customers, staff, appointments, sales, etc.) use the same meta structure for audit information.
+
+### 2. Human-Readable Audit Trail
+Response includes both username (human-readable) and user_id (for system operations):
+- `meta.created_by`: "admin" (easy to understand)
+- `meta.created_by_id`: "usr_01HZQX..." (for system lookups)
+
+### 3. Clean API Responses
+Audit fields are grouped under 'meta', reducing clutter in the main response body.
+
+### 4. Reusability
+Utilities are in `app/core/utils.py` and can be used across all modules without duplication.
+
+### 5. Type Safety
+Pydantic schemas ensure correct structure and validation.
+
+### 6. Database Flexibility
+Database documents maintain flat structure (good for querying), while API responses use nested structure (good for clarity).
+
+## Implementation Pattern Summary
+
+For any new module or entity, follow this pattern:
+
+### Step 1: Update Model
+```python
+# Add audit fields to MongoDB model
+created_by: str
+created_by_username: Optional[str]
+created_at: datetime
+updated_by: Optional[str]
+updated_by_username: Optional[str]
+updated_at: datetime
+```
+
+### Step 2: Update Schemas
+```python
+# Create schema: add created_by_username
+class EntityCreate(BaseModel):
+    created_by_username: Optional[str] = None
+
+# Update schema: add updated_by_username
+class EntityUpdate(BaseModel):
+    updated_by_username: Optional[str] = None
+
+# Response schema: replace timestamps with meta
+class EntityResponse(BaseModel):
+    meta: Dict[str, Any]  # Instead of created_at, updated_at
+```
+
+### Step 3: Update Service
+```python
+from app.core.utils import format_meta_field, extract_audit_fields
+
+# In create method
+async def create_entity(payload, user_id, username):
+    data = payload.dict()
+    audit_fields = extract_audit_fields(user_id, username, is_update=False)
+    data.update(audit_fields)
+    # ... insert into DB ...
+    formatted = format_meta_field(data)
+    return EntityResponse(**formatted)
+
+# In update method
+async def update_entity(entity_id, payload, user_id, username):
+    update_data = payload.dict(exclude_unset=True)
+    audit_fields = extract_audit_fields(user_id, username, is_update=True)
+    update_data.update(audit_fields)
+    # ... update in DB ...
+    doc = await get_entity(entity_id)
+    formatted = format_meta_field(doc)
+    return EntityResponse(**formatted)
+
+# In get/list methods
+async def get_entity(entity_id):
+    doc = await db.find_one({"entity_id": entity_id})
+    formatted = format_meta_field(doc)
+    return EntityResponse(**formatted)
+```
+
+### Step 4: Update Controller
+```python
+# Pass user info from token to service
+@router.post("/")
+async def create_entity(
+    payload: EntityCreate,
+    current_user: TokenUser = Depends(get_current_user)
+):
+    payload.created_by_username = current_user.username
+    return await EntityService.create_entity(
+        payload,
+        user_id=current_user.user_id,
+        username=current_user.username
+    )
+
+@router.put("/{entity_id}")
+async def update_entity(
+    entity_id: str,
+    payload: EntityUpdate,
+    current_user: TokenUser = Depends(get_current_user)
+):
+    payload.updated_by_username = current_user.username
+    return await EntityService.update_entity(
+        entity_id,
+        payload,
+        user_id=current_user.user_id,
+        username=current_user.username
+    )
+```
+
+## Testing Checklist
+
+For each updated module:
+
+- [ ] Test POST (create) endpoint - verify meta.created_by and meta.created_at
+- [ ] Test GET (retrieve) endpoint - verify meta structure
+- [ ] Test PUT/PATCH (update) endpoint - verify meta.updated_by and meta.updated_at
+- [ ] Test DELETE (soft delete) endpoint - verify meta.updated_by
+- [ ] Test list endpoint without projection - verify meta in each item
+- [ ] Test list endpoint with projection - verify raw fields returned
+- [ ] Verify database documents maintain flat structure
+- [ ] Verify API responses use nested meta structure
+
+## Migration Notes
+
+### Backward Compatibility
+
+- **Database**: Existing documents without audit fields will still work (fields are optional in get operations)
+- **API**: Responses now include 'meta' instead of top-level created_at/updated_at
+- **Clients**: Frontend/mobile apps will need updates to read from meta object
+
+### Database Migration
+
+No migration needed for existing documents. New documents will have audit fields, old documents will work without them (format_meta_field handles missing fields gracefully).
+
+### Rollback Plan
+
+If needed, can revert by:
+1. Removing meta field from response schemas
+2. Adding back created_at/updated_at to response schemas
+3. Removing format_meta_field calls from services
+4. Removing extract_audit_fields calls from create/update operations
+
+## Future Enhancements
+
+### Potential Additions
+
+1. **Deletion Audit**: Add deleted_by, deleted_by_username, deleted_at for hard deletes
+2. **Version Tracking**: Add version number to meta for optimistic locking
+3. **Change History**: Store full change history in separate collection
+4. **IP Address Tracking**: Add request_ip to audit fields
+5. **Reason Tracking**: Add change_reason field for important updates
+
+### Other Modules to Update
+
+- **Appointments Module**: Apply same pattern
+- **Sales Module**: Apply same pattern
+- **Wallet Module**: Apply same pattern
+- **Inventory Module** (if exists): Apply same pattern
+
+## Related Documentation
+
+- [Core README](/app/core/README.md) - Detailed usage guide for core schemas and utilities
+- [SCM Meta Implementation](../../cuatrolabs-scm-ms/TRANSPORTS_META_IMPLEMENTATION.md) - Similar implementation in SCM microservice
+
+## Contact
+
+For questions or issues with this implementation, contact the backend team.
+
+## Version History
+
+- **v1.0** (December 2024): Initial implementation for Customers and Staff modules

app/catalogue_services/schemas/schema.py

@@ -1,7 +1,7 @@
 """
 Pydantic schemas for Catalogue Services API.
 """
-from typing import Optional, List, Union
+from typing import Optional, List, Union, Dict, Any
 from pydantic import BaseModel, Field, constr
 from datetime import datetime
 
@@ -23,6 +23,7 @@ class CreateServiceRequest(BaseModel):
     description: Optional[str] = None
     duration_mins: int = Field(..., ge=1)
     pricing: Pricing
+    created_by_username: Optional[str] = Field(None, description="Username who created - will be set from token")
 
 class UpdateServiceRequest(BaseModel):
     name: Optional[str] = Field(None, min_length=1, max_length=200)
@@ -32,6 +33,7 @@ class UpdateServiceRequest(BaseModel):
     description: Optional[str] = None
     duration_mins: Optional[int] = Field(None, ge=1)
     pricing: Optional[Pricing] = None
+    updated_by_username: Optional[str] = Field(None, description="Username who updated - will be set from token")
 
 
 class UpdateStatusRequest(BaseModel):
@@ -48,8 +50,7 @@ class ServiceResponse(BaseModel):
     pricing: dict
     status: str
     sort_order: int = 0
-    created_at: datetime
-    updated_at: datetime
+    meta: Dict[str, Any] = Field(..., description="Audit information (created_by, created_at, updated_by, updated_at)")
 
 class ListServicesRequest(BaseModel):
     merchant_id: Optional[str] = None

app/core/README.md

@@ -0,0 +1,316 @@
+# Core Schemas and Utilities
+
+This directory contains reusable schemas and utility functions that can be used across all modules in the POS microservice.
+
+## Files
+
+### `schemas.py`
+Contains common Pydantic schemas for consistent API responses.
+
+### `utils.py`
+Contains utility functions for common operations like formatting meta fields.
+
+## Available Schemas
+
+### 1. MetaSchema
+
+Groups audit information in a consistent structure across all entities.
+
+**Usage in Response Schemas:**
+
+```python
+from app.core.schemas import MetaSchema
+
+class CustomerResponse(BaseModel):
+    customer_id: str
+    name: str
+    # ... other fields
+    meta: Dict[str, Any]  # Will contain MetaSchema structure
+```
+
+**Fields:**
+- `created_by`: Username who created the record (human-readable)
+- `created_by_id`: UUID who created the record
+- `created_at`: Creation timestamp
+- `updated_by`: Username who last updated (optional)
+- `updated_by_id`: UUID who last updated (optional)
+- `updated_at`: Last update timestamp (optional)
+
+**Example Response:**
+```json
+{
+  "customer_id": "cus_123",
+  "name": "John Doe",
+  "meta": {
+    "created_by": "admin",
+    "created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
+    "created_at": "2023-01-10T08:00:00Z",
+    "updated_by": "manager",
+    "updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
+    "updated_at": "2024-11-24T10:00:00Z"
+  }
+}
+```
+
+### 2. PaginationMeta
+
+Provides pagination information for list endpoints.
+
+### 3. ErrorDetail
+
+Standard error detail structure for validation errors.
+
+### 4. SuccessResponse
+
+Standard success response for operations without data.
+
+### 5. StatusResponse
+
+Standard status response for operations with optional data payload.
+
+### 6. BaseLazyFetchSchema
+
+Base schema for lazy-fetch list requests with pagination and projection support.
+
+## Available Utilities
+
+### 1. format_meta_field()
+
+Transforms database documents to API response format by grouping audit fields under 'meta'.
+
+**Usage in Service Layer:**
+
+```python
+from app.core.utils import format_meta_field
+
+class CustomerService:
+    @staticmethod
+    async def get_customer(customer_id: str) -> CustomerResponse:
+        # Fetch from database
+        customer = await db.customers.find_one({"customer_id": customer_id})
+        
+        # Format with meta field
+        formatted = format_meta_field(customer)
+        
+        # Return response
+        return CustomerResponse(**formatted)
+```
+
+**What it does:**
+- Extracts audit fields from document
+- Creates 'meta' object with proper field mapping
+- Removes audit fields from top level
+- Returns formatted document
+
+**Field Mapping:**
+```
+DB Field              → Response Field
+----------------        ---------------
+created_by (UUID)     → meta.created_by_id
+created_by_username   → meta.created_by
+created_at            → meta.created_at
+updated_by (UUID)     → meta.updated_by_id
+updated_by_username   → meta.updated_by
+updated_at            → meta.updated_at
+```
+
+### 2. extract_audit_fields()
+
+Creates audit field dictionaries for database operations.
+
+**Usage for Creation:**
+
+```python
+from app.core.utils import extract_audit_fields
+
+# When creating a new record
+audit_fields = extract_audit_fields(
+    user_id=current_user.user_id,
+    username=current_user.username,
+    is_update=False
+)
+
+customer_data = {
+    "customer_id": "cus_123",
+    "name": "John Doe",
+    **audit_fields  # Adds created_by, created_by_username, created_at
+}
+```
+
+**Usage for Updates:**
+
+```python
+# When updating a record
+audit_fields = extract_audit_fields(
+    user_id=current_user.user_id,
+    username=current_user.username,
+    is_update=True
+)
+
+update_data = {
+    "name": "New Name",
+    **audit_fields  # Adds updated_by, updated_by_username, updated_at
+}
+```
+
+### 3. normalize_uuid_fields()
+
+Converts UUID objects to strings for JSON serialization.
+
+## Implementation Guide for New Modules
+
+### Step 1: Update Model
+
+Add audit fields to your MongoDB model:
+
+```python
+from pydantic import BaseModel, Field
+from datetime import datetime
+from typing import Optional
+
+class CustomerModel(BaseModel):
+    customer_id: str
+    name: str
+    # ... other fields
+    
+    # Audit fields
+    created_by: str = Field(..., description="User ID who created")
+    created_by_username: Optional[str] = Field(None, description="Username who created")
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_by: Optional[str] = Field(None, description="User ID who updated")
+    updated_by_username: Optional[str] = Field(None, description="Username who updated")
+    updated_at: Optional[datetime] = Field(None)
+```
+
+### Step 2: Update Response Schema
+
+Import MetaSchema and use it in your response:
+
+```python
+from typing import Dict, Any
+from pydantic import BaseModel
+from app.core.schemas import MetaSchema
+
+class CustomerResponse(BaseModel):
+    customer_id: str
+    name: str
+    # ... other fields
+    meta: Dict[str, Any]  # Contains MetaSchema structure
+```
+
+### Step 3: Update Service Layer
+
+Use the utility functions in your service:
+
+```python
+from app.core.utils import format_meta_field, extract_audit_fields
+
+class CustomerService:
+    @staticmethod
+    async def create_customer(payload: CustomerCreate, current_user: TokenUser):
+        # Add audit fields for creation
+        audit_fields = extract_audit_fields(
+            user_id=current_user.user_id,
+            username=current_user.username,
+            is_update=False
+        )
+        
+        customer_data = {
+            **payload.dict(),
+            **audit_fields
+        }
+        
+        # Insert into database
+        await db.customers.insert_one(customer_data)
+        
+        # Format response with meta
+        formatted = format_meta_field(customer_data)
+        return CustomerResponse(**formatted)
+    
+    @staticmethod
+    async def update_customer(
+        customer_id: str, 
+        payload: CustomerUpdate,
+        current_user: TokenUser
+    ):
+        # Add audit fields for update
+        audit_fields = extract_audit_fields(
+            user_id=current_user.user_id,
+            username=current_user.username,
+            is_update=True
+        )
+        
+        update_data = {
+            **payload.dict(exclude_unset=True),
+            **audit_fields
+        }
+        
+        # Update in database
+        await db.customers.update_one(
+            {"customer_id": customer_id},
+            {"$set": update_data}
+        )
+        
+        # Fetch and format response
+        customer = await db.customers.find_one({"customer_id": customer_id})
+        formatted = format_meta_field(customer)
+        return CustomerResponse(**formatted)
+    
+    @staticmethod
+    async def get_customer(customer_id: str):
+        customer = await db.customers.find_one({"customer_id": customer_id})
+        if not customer:
+            raise HTTPException(status_code=404, detail="Customer not found")
+        
+        # Format with meta
+        formatted = format_meta_field(customer)
+        return CustomerResponse(**formatted)
+```
+
+### Step 4: Update Controller
+
+Ensure controllers pass both user_id and username:
+
+```python
+@router.post("/")
+async def create_customer(
+    payload: CustomerCreate,
+    current_user: TokenUser = Depends(get_current_user)
+):
+    return await CustomerService.create_customer(payload, current_user)
+
+@router.patch("/{customer_id}")
+async def update_customer(
+    customer_id: str,
+    payload: CustomerUpdate,
+    current_user: TokenUser = Depends(get_current_user)
+):
+    return await CustomerService.update_customer(
+        customer_id, 
+        payload, 
+        current_user
+    )
+```
+
+## Benefits
+
+1. **Consistency**: All entities use the same meta structure
+2. **Reusability**: Write once, use everywhere
+3. **Maintainability**: Changes to meta structure happen in one place
+4. **Type Safety**: Pydantic validation ensures correct structure
+5. **Documentation**: Self-documenting with clear field descriptions
+
+## Migration Checklist
+
+When migrating an existing module to use MetaSchema:
+
+- [ ] Add audit fields to model (if not present)
+- [ ] Import MetaSchema in response schema
+- [ ] Update response schema to use `meta: Dict[str, Any]`
+- [ ] Import utility functions in service
+- [ ] Update create method to use `extract_audit_fields(is_update=False)`
+- [ ] Update update method to use `extract_audit_fields(is_update=True)`
+- [ ] Update all methods returning responses to use `format_meta_field()`
+- [ ] Update controller to pass `current_user` to service methods
+- [ ] Test all CRUD operations
+- [ ] Update API documentation

app/core/schemas.py

@@ -0,0 +1,142 @@
+"""
+Core/Common Pydantic schemas used across multiple modules.
+These schemas provide reusable components for consistent API responses.
+"""
+from typing import Optional, Any, Dict, List
+from datetime import datetime
+from pydantic import BaseModel, Field
+
+
+class MetaSchema(BaseModel):
+    """
+    Audit and metadata information for tracking record lifecycle.
+    
+    This schema groups all audit-related fields in a consistent structure
+    that can be used across all entities (customers, staff, appointments, etc.).
+    
+    Fields:
+        created_by: Username of the user who created the record (human-readable)
+        created_by_id: UUID of the user who created the record
+        created_at: Timestamp when the record was created
+        updated_by: Username of the user who last updated the record (optional)
+        updated_by_id: UUID of the user who last updated the record (optional)
+        updated_at: Timestamp when the record was last updated (optional)
+    """
+    created_by: str = Field(..., description="Username who created this record")
+    created_by_id: str = Field(..., description="User ID (UUID) who created this record")
+    created_at: datetime = Field(..., description="Timestamp when record was created")
+    updated_by: Optional[str] = Field(None, description="Username who last updated this record")
+    updated_by_id: Optional[str] = Field(None, description="User ID (UUID) who last updated this record")
+    updated_at: Optional[datetime] = Field(None, description="Timestamp when record was last updated")
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "created_by": "admin",
+                "created_by_id": "usr_01HZQX5K3N2P8R6T4V9W",
+                "created_at": "2023-01-10T08:00:00Z",
+                "updated_by": "manager",
+                "updated_by_id": "usr_01HZQX5K3N2P8R6T4V9X",
+                "updated_at": "2024-11-24T10:00:00Z"
+            }
+        }
+
+
+class PaginationMeta(BaseModel):
+    """
+    Pagination metadata for list endpoints.
+    
+    Provides information about the current page, total records, and pagination state.
+    """
+    total: int = Field(..., description="Total number of records matching the query")
+    skip: int = Field(..., description="Number of records skipped (offset)")
+    limit: int = Field(..., description="Maximum number of records returned")
+    has_more: bool = Field(..., description="Whether there are more records available")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "total": 150,
+                "skip": 0,
+                "limit": 100,
+                "has_more": True
+            }
+        }
+
+
+class ErrorDetail(BaseModel):
+    """
+    Detailed error information for validation and business logic errors.
+    """
+    field: str = Field(..., description="Field name that caused the error")
+    message: str = Field(..., description="Human-readable error message")
+    code: Optional[str] = Field(None, description="Machine-readable error code")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "field": "email",
+                "message": "Email address is already registered",
+                "code": "DUPLICATE_EMAIL"
+            }
+        }
+
+
+class SuccessResponse(BaseModel):
+    """
+    Standard success response for operations that don't return data.
+    """
+    success: bool = Field(True, description="Operation success status")
+    message: str = Field(..., description="Success message")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "success": True,
+                "message": "Operation completed successfully"
+            }
+        }
+
+
+class StatusResponse(BaseModel):
+    """
+    Standard status response for operations with optional data payload.
+    Used for API endpoints that return status with additional data.
+    """
+    success: bool = Field(..., description="Operation success status")
+    message: str = Field(..., description="Status or success message")
+    data: Optional[Dict[str, Any]] = Field(None, description="Optional data payload")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "success": True,
+                "message": "Operation completed successfully",
+                "data": {
+                    "count": 10,
+                    "items": []
+                }
+            }
+        }
+
+
+class BaseLazyFetchSchema(BaseModel):
+    """
+    Base schema for lazy-fetch list requests with pagination and projection support.
+    Provides common fields for listing endpoints with filtering, pagination, and field projection.
+    """
+    skip: int = Field(0, ge=0, description="Number of records to skip (pagination offset)")
+    limit: int = Field(100, ge=1, le=500, description="Maximum number of records to return")
+    projection_list: Optional[List[str]] = Field(
+        None, 
+        description="List of fields to include in response (omit for all fields)"
+    )
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "skip": 0,
+                "limit": 100,
+                "projection_list": ["id", "name", "status"]
+            }
+        }

app/core/utils.py

@@ -0,0 +1,160 @@
+"""
+Core utility functions for common operations across modules.
+"""
+from typing import Dict, Any, Optional
+from datetime import datetime
+
+
+def format_meta_field(document: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Format a document by grouping audit fields under 'meta'.
+    
+    This utility transforms database documents to API response format by:
+    1. Extracting audit fields (created_by, created_by_username, etc.)
+    2. Creating a 'meta' object with proper field mapping
+    3. Removing audit fields from the top level
+    
+    Field Mapping:
+        DB Field              → Response Field
+        ----------------        ---------------
+        created_by (UUID)     → meta.created_by_id
+        created_by_username   → meta.created_by
+        created_at            → meta.created_at
+        updated_by (UUID)     → meta.updated_by_id
+        updated_by_username   → meta.updated_by
+        updated_at            → meta.updated_at
+    
+    Args:
+        document: Raw document from database with audit fields
+        
+    Returns:
+        Formatted document with 'meta' field containing audit information
+        
+    Example:
+        >>> doc = {
+        ...     "id": "123",
+        ...     "name": "Test",
+        ...     "created_by": "usr_admin",
+        ...     "created_by_username": "admin",
+        ...     "created_at": "2024-01-01T00:00:00Z"
+        ... }
+        >>> formatted = format_meta_field(doc)
+        >>> formatted
+        {
+            "id": "123",
+            "name": "Test",
+            "meta": {
+                "created_by": "admin",
+                "created_by_id": "usr_admin",
+                "created_at": "2024-01-01T00:00:00Z"
+            }
+        }
+    """
+    # Create meta object from audit fields
+    meta = {
+        "created_by": document.get("created_by_username") or "Unknown",
+        "created_by_id": str(document.get("created_by", "")),
+        "created_at": document.get("created_at"),
+    }
+    
+    # Add optional updated fields if they exist
+    if "updated_by" in document and document["updated_by"]:
+        meta["updated_by_id"] = str(document["updated_by"])
+        meta["updated_by"] = document.get("updated_by_username") or "Unknown"
+    
+    if "updated_at" in document and document["updated_at"]:
+        meta["updated_at"] = document["updated_at"]
+    
+    # Create a copy of document without the audit fields
+    audit_fields = [
+        "created_by", 
+        "created_at", 
+        "created_by_username", 
+        "updated_by", 
+        "updated_by_username", 
+        "updated_at"
+    ]
+    formatted = {k: v for k, v in document.items() if k not in audit_fields}
+    
+    # Add the meta field
+    formatted["meta"] = meta
+    
+    return formatted
+
+
+def extract_audit_fields(
+    user_id: str,
+    username: Optional[str] = None,
+    is_update: bool = False
+) -> Dict[str, Any]:
+    """
+    Create audit field dictionaries for database operations.
+    
+    Generates the appropriate audit fields for create or update operations,
+    including both user ID (for queries) and username (for display).
+    
+    Args:
+        user_id: User UUID performing the operation
+        username: Username for display (optional, defaults to user_id if not provided)
+        is_update: True for updates, False for creation
+        
+    Returns:
+        Dictionary with audit fields ready for database insertion
+        
+    Example:
+        >>> # For creation
+        >>> audit = extract_audit_fields("usr_123", "admin", is_update=False)
+        >>> audit
+        {
+            "created_by": "usr_123",
+            "created_by_username": "admin",
+            "created_at": datetime(...)
+        }
+        
+        >>> # For updates
+        >>> audit = extract_audit_fields("usr_456", "manager", is_update=True)
+        >>> audit
+        {
+            "updated_by": "usr_456",
+            "updated_by_username": "manager",
+            "updated_at": datetime(...)
+        }
+    """
+    now = datetime.utcnow()
+    
+    if is_update:
+        return {
+            "updated_by": user_id,
+            "updated_by_username": username or user_id,
+            "updated_at": now
+        }
+    else:
+        return {
+            "created_by": user_id,
+            "created_by_username": username or user_id,
+            "created_at": now
+        }
+
+
+def normalize_uuid_fields(document: Dict[str, Any], fields: list[str]) -> Dict[str, Any]:
+    """
+    Convert UUID objects to strings for JSON serialization.
+    
+    Args:
+        document: Document with potential UUID fields
+        fields: List of field names that might contain UUIDs
+        
+    Returns:
+        Document with UUID fields converted to strings
+        
+    Example:
+        >>> from uuid import UUID
+        >>> doc = {"user_id": UUID("123e4567-e89b-12d3-a456-426614174000")}
+        >>> normalized = normalize_uuid_fields(doc, ["user_id"])
+        >>> normalized["user_id"]
+        '123e4567-e89b-12d3-a456-426614174000'
+    """
+    for field in fields:
+        if field in document and document[field] is not None:
+            document[field] = str(document[field])
+    return document

app/customers/controllers/router.py

@@ -40,7 +40,15 @@ async def create_customer(
         if not current_user.merchant_id:
             raise HTTPException(status_code=400, detail="merchant_id must be available in token")
         
-        customer = await CustomerService.create_customer(payload, current_user.merchant_id)
+        # Pass username to service for audit trail
+        payload.created_by_username = current_user.username
+        
+        customer = await CustomerService.create_customer(
+            payload, 
+            current_user.merchant_id,
+            current_user.user_id,
+            current_user.username
+        )
         
         logger.info(
             "Customer created successfully",
@@ -200,11 +208,16 @@ async def update_customer(
     current_user: TokenUser = Depends(require_pos_permission("customers", "update"))
 ) -> CustomerResponse:
     try:
+        # Set username for audit trail
+        payload.updated_by_username = current_user.username
+        
         customer = await CustomerService.update_customer(
             customer_id=customer_id,
             payload=payload,
             merchant_id=current_user.merchant_id,
-            merchant_type=current_user.merchant_type
+            merchant_type=current_user.merchant_type,
+            user_id=current_user.user_id,
+            username=current_user.username
         )
         
         logger.info(
@@ -251,7 +264,9 @@ async def delete_customer(
         await CustomerService.delete_customer(
             customer_id=customer_id,
             merchant_id=current_user.merchant_id,
-            merchant_type=current_user.merchant_type
+            merchant_type=current_user.merchant_type,
+            user_id=current_user.user_id,
+            username=current_user.username
         )
         
         logger.info(

app/customers/models/model.py

@@ -14,7 +14,13 @@ class CustomerModel(BaseModel):
     email: Optional[EmailStr] = None
     notes: Optional[str] = Field(None, max_length=500)
     status: str = Field(default="active")
+    
+    # Audit fields
+    created_by: str = Field(..., description="User ID who created")
+    created_by_username: Optional[str] = Field(None, description="Username who created")
     created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_by: Optional[str] = Field(None, description="User ID who updated")
+    updated_by_username: Optional[str] = Field(None, description="Username who updated")
     updated_at: datetime = Field(default_factory=datetime.utcnow)
 
     class Config:

app/customers/schemas/schema.py

@@ -2,7 +2,7 @@
 Pydantic schemas for POS customers.
 """
 from datetime import date, datetime
-from typing import Optional, List, Union
+from typing import Optional, List, Union, Dict, Any
 from pydantic import BaseModel, Field, EmailStr, field_validator
 import re
 
@@ -76,6 +76,7 @@ class CustomerBase(BaseModel):
 class CustomerCreate(CustomerBase):
     """Schema for creating a customer."""
     merchant_id: Optional[str] = Field(None, description="Merchant identifier (UUID) - will be overridden by token")
+    created_by_username: Optional[str] = Field(None, description="Username who created - will be set from token")
     
     @field_validator("phone")
     @classmethod
@@ -122,6 +123,8 @@ class CustomerUpdate(BaseModel):
     referral_discount_used: Optional[float] = None
     referred_by: Optional[str] = None
     
+    updated_by_username: Optional[str] = Field(None, description="Username who updated - will be set from token")
+    
     @field_validator("phone")
     @classmethod
     def validate_phone(cls, v: Optional[str]) -> Optional[str]:
@@ -151,8 +154,7 @@ class CustomerUpdate(BaseModel):
         
 class CustomerResponse(CustomerBase):
     customer_id: str = Field(..., description="Customer identifier (UUID)")
-    created_at: datetime
-    updated_at: datetime
+    meta: Dict[str, Any] = Field(..., description="Audit information (created_by, created_at, updated_by, updated_at)")
 
     class Config:
         from_attributes = True

app/customers/services/service.py

@@ -10,6 +10,7 @@ from app.utils.utils import normalize_dates
 from sqlalchemy import text
 
 from app.core.logging import get_logger
+from app.core.utils import format_meta_field, extract_audit_fields
 from app.nosql import get_database
 from app.constants.collections import POS_CUSTOMERS_COLLECTION
 from app.sql import get_postgres_session
@@ -92,22 +93,31 @@ class CustomerService:
             )
 
     @classmethod
-    async def create_customer(cls, payload: CustomerCreate, token_merchant_id: Optional[str] = None) -> CustomerResponse:
+    async def create_customer(cls, payload: CustomerCreate, token_merchant_id: Optional[str] = None, user_id: Optional[str] = None, username: Optional[str] = None) -> CustomerResponse:
         now = datetime.utcnow()
         customer_id = cls._generate_customer_id()
 
         doc = payload.model_dump()
         doc["customer_id"] = customer_id
-        doc["created_at"] = now
-        doc["updated_at"] = now
         
         # Override merchant_id with UUID format from token if provided
         if token_merchant_id:
             doc["merchant_id"] = token_merchant_id
+        
+        # Add audit fields
+        audit_fields = extract_audit_fields(
+            user_id=user_id or "system",
+            username=username or "system",
+            is_update=False
+        )
+        doc.update(audit_fields)
 
         data = normalize_dates(doc)
         await cls._collection().insert_one(data)
-        customer = CustomerResponse(**data)
+        
+        # Format response with meta
+        formatted = format_meta_field(data)
+        customer = CustomerResponse(**formatted)
 
         try:
             await cls._sync_to_postgres(customer, token_merchant_id)
@@ -142,7 +152,10 @@ class CustomerService:
         doc = await cls._collection().find_one(query)
         if not doc:
             return None
-        return CustomerResponse(**doc)
+        
+        # Format with meta
+        formatted = format_meta_field(doc)
+        return CustomerResponse(**formatted)
 
     @classmethod
     async def list_customers(
@@ -187,10 +200,11 @@ class CustomerService:
             docs = await cursor.to_list(length=limit)
             return docs, total
         else:
-            # Return CustomerResponse objects when no projection
+            # Return CustomerResponse objects when no projection, formatted with meta
             items: List[CustomerResponse] = []
             async for doc in cursor:
-                items.append(CustomerResponse(**doc))
+                formatted = format_meta_field(doc)
+                items.append(CustomerResponse(**formatted))
             return items, total
 
     @classmethod
@@ -199,7 +213,9 @@ class CustomerService:
         customer_id: str, 
         payload: CustomerUpdate,
         merchant_id: Optional[str] = None,
-        merchant_type: Optional[str] = None
+        merchant_type: Optional[str] = None,
+        user_id: Optional[str] = None,
+        username: Optional[str] = None
     ) -> CustomerResponse:
         """Update customer with merchant access control."""
         query = {"customer_id": customer_id}
@@ -216,7 +232,13 @@ class CustomerService:
         if not update_data:
             raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No fields to update")
 
-        update_data["updated_at"] = datetime.utcnow()
+        # Add audit fields for update
+        audit_fields = extract_audit_fields(
+            user_id=user_id or "system",
+            username=username or "system",
+            is_update=True
+        )
+        update_data.update(audit_fields)
 
         await cls._collection().update_one(
             {"customer_id": customer_id},
@@ -224,7 +246,8 @@ class CustomerService:
         )
 
         updated = await cls._collection().find_one({"customer_id": customer_id})
-        customer = CustomerResponse(**updated)
+        formatted = format_meta_field(updated)
+        customer = CustomerResponse(**formatted)
 
         try:
             await cls._sync_to_postgres(customer, merchant_id)
@@ -247,7 +270,9 @@ class CustomerService:
         cls, 
         customer_id: str,
         merchant_id: Optional[str] = None,
-        merchant_type: Optional[str] = None
+        merchant_type: Optional[str] = None,
+        user_id: Optional[str] = None,
+        username: Optional[str] = None
     ) -> None:
         """Delete (soft) customer with merchant access control."""
         query = {"customer_id": customer_id}
@@ -260,14 +285,25 @@ class CustomerService:
         if not existing:
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Customer not found")
 
-        # Soft delete: set status inactive and updated_at
+        # Soft delete: set status inactive and add audit fields
+        audit_fields = extract_audit_fields(
+            user_id=user_id or "system",
+            username=username or "system",
+            is_update=True
+        )
+        update_data = {
+            "status": "inactive",
+            **audit_fields
+        }
+        
         await cls._collection().update_one(
             {"customer_id": customer_id},
-            {"$set": {"status": "inactive", "updated_at": datetime.utcnow()}}
+            {"$set": update_data}
         )
 
         updated = await cls._collection().find_one({"customer_id": customer_id})
-        customer = CustomerResponse(**updated)
+        formatted = format_meta_field(updated)
+        customer = CustomerResponse(**formatted)
 
         try:
             await cls._sync_to_postgres(customer, merchant_id)

app/staff/controllers/router.py

@@ -55,8 +55,13 @@ async def create_staff(
             raise HTTPException(status_code=400, detail="merchant_id not found in token")
         
         payload.merchant_id = current_user.merchant_id
+        payload.created_by_username = current_user.username
         
-        result = await StaffService.create_staff(payload)
+        result = await StaffService.create_staff(
+            payload,
+            user_id=current_user.user_id,
+            username=current_user.username
+        )
         
         logger.info(
             "Staff member created",
@@ -222,8 +227,17 @@ async def update_staff(
     try:
         if not current_user.merchant_id:
             raise HTTPException(status_code=400, detail="merchant_id not found in token")
-            
-        result = await StaffService.update_staff(staff_id, current_user.merchant_id, payload)
+        
+        # Set username for audit trail
+        payload.updated_by_username = current_user.username
+        
+        result = await StaffService.update_staff(
+            staff_id, 
+            current_user.merchant_id, 
+            payload,
+            user_id=current_user.user_id,
+            username=current_user.username
+        )
         
         logger.info(
             "Staff member updated",
@@ -274,8 +288,17 @@ async def update_staff_status(
         if not current_user.merchant_id:
             raise HTTPException(status_code=400, detail="merchant_id not found in token")
         
-        update_payload = StaffUpdateSchema(status=payload.status)
-        result = await StaffService.update_staff(staff_id, current_user.merchant_id, update_payload)
+        update_payload = StaffUpdateSchema(
+            status=payload.status,
+            updated_by_username=current_user.username
+        )
+        result = await StaffService.update_staff(
+            staff_id, 
+            current_user.merchant_id, 
+            update_payload,
+            user_id=current_user.user_id,
+            username=current_user.username
+        )
         
         logger.info(
             "Staff status updated",

app/staff/models/staff_model.py

@@ -42,7 +42,12 @@ class StaffModel(BaseModel):
     
     working_hours: List[WorkingHours] = Field(default_factory=list, description="Weekly working schedule")
     
+    # Audit fields
+    created_by: str = Field(..., description="User ID who created")
+    created_by_username: Optional[str] = Field(None, description="Username who created")
     created_at: datetime = Field(default_factory=datetime.utcnow, description="Creation timestamp")
+    updated_by: Optional[str] = Field(None, description="User ID who updated")
+    updated_by_username: Optional[str] = Field(None, description="Username who updated")
     updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update timestamp")
     
     # Optional fields

app/staff/schemas/staff_schema.py

@@ -30,6 +30,7 @@ class StaffCreateSchema(BaseModel):
     working_hours: Optional[List[dict]] = None  
     category: Optional[str] = None  
     is_deleted: bool = Field(default=False)
+    created_by_username: Optional[str] = Field(None, description="Username who created - will be set from token")
     @field_validator('phone')
     @classmethod
     def validate_phone(cls, v):
@@ -63,7 +64,8 @@ class StaffUpdateSchema(BaseModel):
     status: Optional[str] = Field(None, pattern="^(active|inactive|on_leave|suspended|terminated)$")
     skills: Optional[List[str]] = None
     working_hours: Optional[List[dict]] = None  
-    category: Optional[str] = None  
+    category: Optional[str] = None
+    updated_by_username: Optional[str] = Field(None, description="Username who updated - will be set from token")  
     @field_validator('phone')
     @classmethod
     def validate_phone(cls, v):
@@ -103,8 +105,7 @@ class StaffResponseSchema(BaseModel):
     status: str
     working_hours: Optional[List[dict]] = None
     category: Optional[str] = None
-    created_at: Optional[datetime] = None
-    updated_at: Optional[datetime] = None
+    meta: dict = Field(..., description="Audit information (created_by, created_at, updated_by, updated_at)")
     
     class Config:
         from_attributes = True

app/staff/services/staff_service.py

@@ -7,6 +7,7 @@ from typing import Optional, List, Dict, Any, Union
 from uuid import UUID, uuid4
 from fastapi import HTTPException, status
 from app.core.logging import get_logger
+from app.core.utils import format_meta_field, extract_audit_fields
 import secrets
 from sqlalchemy import text
 
@@ -93,12 +94,14 @@ class StaffService:
     """Service class for staff operations."""
 
     @staticmethod
-    async def create_staff(payload: StaffCreateSchema) -> StaffResponseSchema:
+    async def create_staff(payload: StaffCreateSchema, user_id: Optional[str] = None, username: Optional[str] = None) -> StaffResponseSchema:
         """
         Create a new staff member.
         
         Args:
             payload: Staff creation data
+            user_id: User ID from token for audit trail
+            username: Username from token for audit trail
             
         Returns:
             Created staff response
@@ -141,15 +144,20 @@ class StaffService:
 
             # Generate staff ID
             staff_id = generate_uuid()
-            user_id = generate_uuid()
+            user_id_gen = generate_uuid()
             
             # Create staff model
-            now = datetime.utcnow()
             staff_data = payload.model_dump(by_alias=True)
             staff_data["staff_id"] = staff_id
-            staff_data["user_id"] = user_id
-            staff_data["created_at"] = now
-            staff_data["updated_at"] = now
+            staff_data["user_id"] = user_id_gen
+            
+            # Add audit fields
+            audit_fields = extract_audit_fields(
+                user_id=user_id or "system",
+                username=username or "system",
+                is_update=False
+            )
+            staff_data.update(audit_fields)
             
             # Insert into database
             await get_database()[POS_STAFF_COLLECTION].insert_one(staff_data)
@@ -177,7 +185,7 @@ class StaffService:
                 user_payload = SystemUserCreateSchema(
                     merchant_id=payload.merchant_id,
                     staff_id=staff_id,
-                    user_id=user_id,
+                    user_id=user_id_gen,
                     name=payload.name,
                     email=payload.email,
                     phone=payload.phone,
@@ -196,8 +204,10 @@ class StaffService:
                     },
                     exc_info=True
                 )
-            # Return response
-            return StaffResponseSchema(**staff_data)
+            
+            # Format response with meta
+            formatted = format_meta_field(staff_data)
+            return StaffResponseSchema(**formatted)
             
         except HTTPException:
             raise
@@ -221,10 +231,12 @@ class StaffService:
         """Get staff by ID."""
         try:
             staff = await get_database()[POS_STAFF_COLLECTION].find_one({"staff_id": str(staff_id), "merchant_id": str(merchant_id)})
-            print(staff)
             if not staff:
                 return None
-            return StaffResponseSchema(**staff)
+            
+            # Format with meta
+            formatted = format_meta_field(staff)
+            return StaffResponseSchema(**formatted)
         except Exception as e:
             logger.error(
                 f"Error fetching staff {staff_id}",
@@ -242,7 +254,7 @@ class StaffService:
             )
 
     @staticmethod
-    async def update_staff(staff_id: UUID, merchant_id: UUID, payload: StaffUpdateSchema) -> StaffResponseSchema:
+    async def update_staff(staff_id: UUID, merchant_id: UUID, payload: StaffUpdateSchema, user_id: Optional[str] = None, username: Optional[str] = None) -> StaffResponseSchema:
         """
         Update staff information.
         
@@ -250,6 +262,8 @@ class StaffService:
             staff_id: Staff ID to update
             merchant_id: Merchant ID
             payload: Update data
+            user_id: User ID from token for audit trail
+            username: Username from token for audit trail
             
         Returns:
             Updated staff response
@@ -284,8 +298,13 @@ class StaffService:
                 detail="No update data provided"
             )
 
-        # Add updated timestamp
-        update_data["updated_at"] = datetime.utcnow()
+        # Add audit fields for update
+        audit_fields = extract_audit_fields(
+            user_id=user_id or "system",
+            username=username or "system",
+            is_update=True
+        )
+        update_data.update(audit_fields)
 
         try:
             # Update in database