feat(staff): Implement PostgreSQL integration for staff management and add validation schemas

app/constants/validation.py

@@ -0,0 +1,36 @@
+"""
+Validation constants and regex patterns.
+"""
+
+# Merchant ID validation
+MERCHANT_ID_REGEX = r"^[a-zA-Z0-9_-]{3,50}$"
+
+# URL validation
+HTTPS_URL_REGEX = r"^https://[^\s]+$"
+
+# Phone validation
+PHONE_E164_REGEX = r"^\+?[1-9]\d{7,14}$"
+
+# Email validation
+EMAIL_REGEX = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
+
+# User ID patterns
+USER_ID_REGEX = r"^usr_[a-zA-Z0-9_-]{10,50}$"
+
+# General alphanumeric patterns
+ALPHANUMERIC_REGEX = r"^[a-zA-Z0-9]+$"
+ALPHANUMERIC_WITH_SPACES_REGEX = r"^[a-zA-Z0-9\s]+$"
+
+# Name validation
+NAME_REGEX = r"^[a-zA-Z\s'-]+$"
+
+# Password strength (at least 8 chars, 1 uppercase, 1 lowercase, 1 number)
+PASSWORD_STRONG_REGEX = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)[a-zA-Z\d@$!%*?&]{8,}$"
+
+# Validation messages
+INVALID_MERCHANT_ID_MSG = "Invalid merchant ID format"
+INVALID_URL_MSG = "Invalid HTTPS URL format"
+INVALID_PHONE_MSG = "Invalid phone number format (E.164)"
+INVALID_EMAIL_MSG = "Invalid email format"
+INVALID_USER_ID_MSG = "Invalid user ID format"
+INVALID_PASSWORD_MSG = "Password must be at least 8 characters with uppercase, lowercase, and number"

app/core/config.py

@@ -19,6 +19,15 @@ class Settings(BaseSettings):
     MONGODB_URI: str = os.getenv("MONGODB_URI", "mongodb://localhost:27017")
     MONGODB_DB_NAME: str = os.getenv("MONGODB_DB_NAME", "pos_db")
 
+    # PostgreSQL Configuration (for trans schema sync)
+    POSTGRES_HOST: str = os.getenv("DB_HOST", "ep-sweet-surf-a1qeduoy.ap-southeast-1.aws.neon.tech")
+    POSTGRES_PORT: int = int(os.getenv("DB_PORT", "5432"))
+    POSTGRES_DB: str = os.getenv("DB_NAME", "cuatrolabs")
+    POSTGRES_USER: str = os.getenv("DB_USER", "trans_owner")
+    POSTGRES_PASSWORD: str = os.getenv("DB_PASSWORD", "BookMyService7")
+    POSTGRES_SSL_MODE: str = os.getenv("DB_SSLMODE", "disable")
+    POSTGRES_URI: Optional[str] = None
+
     # Redis Configuration
     REDIS_HOST: str = os.getenv("REDIS_HOST", "localhost")
     REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379"))
@@ -65,6 +74,14 @@ class Settings(BaseSettings):
         extra="allow",  # allows extra environment variables without error
     )
 
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        # Build PostgreSQL URI from components
+        from urllib.parse import quote_plus
+        if all([self.POSTGRES_USER, self.POSTGRES_PASSWORD, self.POSTGRES_HOST, self.POSTGRES_DB]):
+            protocol = "postgresql+psycopg"
+            self.POSTGRES_URI = f"{protocol}://{self.POSTGRES_USER}:{quote_plus(self.POSTGRES_PASSWORD)}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
+
 
 # Global settings instance
 settings = Settings()

app/main.py

@@ -9,6 +9,7 @@ import logging  # TODO: Uncomment when package is available
 from app.core.config import settings
 
 from app.nosql import connect_to_mongo, close_mongo_connection
+from app.sql import connect_to_postgres, close_postgres_connection
 from app.staff.controllers.router import router as staff_router
 from app.catalogues.controllers.router import router as catalogues_router
 
@@ -41,6 +42,7 @@ async def startup_event():
     """Initialize connections on startup"""
     logger.info("Starting POS Microservice")
     await connect_to_mongo()
+    await connect_to_postgres()
     logger.info("POS Microservice started successfully")
 
 
@@ -49,6 +51,7 @@ async def shutdown_event():
     """Close connections on shutdown"""
     logger.info("Shutting down POS Microservice")
     await close_mongo_connection()
+    await close_postgres_connection()
     logger.info("POS Microservice shut down successfully")
 
 
@@ -63,6 +66,8 @@ async def health_check():
     }
 
 
+# Include routers
+# Authentication is handled by auth-ms, POS just validates tokens
 app.include_router(staff_router, prefix="/api/v1")
 app.include_router(catalogues_router, prefix="/api/v1")
 

app/sql.py

@@ -0,0 +1,119 @@
+"""
+PostgreSQL database connection management for POS microservice.
+Used for syncing staff data to trans.pos_staff_ref table.
+"""
+import logging
+from typing import Optional
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+from contextlib import asynccontextmanager  
+from sqlalchemy import text as sql_text
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+# Global engine instance
+_engine: Optional[AsyncEngine] = None
+_async_session_maker: Optional[sessionmaker] = None
+
+
+async def connect_to_postgres():
+    """Initialize PostgreSQL connection pool."""
+    global _engine, _async_session_maker
+    
+    if not settings.POSTGRES_URI:
+        logger.warning("PostgreSQL URI not configured. Staff sync to trans.pos_staff_ref will be disabled.")
+        return
+    
+    try:
+        _engine = create_async_engine(
+            settings.POSTGRES_URI,
+            pool_pre_ping=True,
+            pool_size=10,
+            max_overflow=20,
+            echo=settings.DEBUG,
+        )
+        
+        _async_session_maker = sessionmaker(
+            _engine,
+            class_=AsyncSession,
+            expire_on_commit=False,
+        )
+        
+        # Test connection
+        async with _engine.begin() as conn:
+            await conn.execute("SELECT 1")
+        
+        logger.info("PostgreSQL connection established successfully")
+        
+    except Exception as e:
+        logger.error(f"Failed to connect to PostgreSQL: {e}")
+        _engine = None
+        _async_session_maker = None
+
+
+async def close_postgres_connection():
+    """Close PostgreSQL connection pool."""
+    global _engine, _async_session_maker
+    
+    if _engine:
+        try:
+            await _engine.dispose()
+            logger.info("PostgreSQL connection closed")
+        except Exception as e:
+            logger.error(f"Error closing PostgreSQL connection: {e}")
+        finally:
+            _engine = None
+            _async_session_maker = None
+
+
+def get_postgres_engine() -> Optional[AsyncEngine]:
+    """Get PostgreSQL engine instance."""
+    return _engine
+
+
+@asynccontextmanager
+async def get_postgres_session():
+    """
+    Get PostgreSQL session context manager.
+    
+    Usage:
+        async with get_postgres_session() as session:
+            await session.execute(...)
+            await session.commit()
+    """
+    if not _async_session_maker:
+        logger.warning("PostgreSQL session maker not initialized")
+        yield None
+        return
+    
+    session = _async_session_maker()
+    try:
+        yield session
+    except Exception as e:
+        await session.rollback()
+        logger.error(f"Session error, rolled back: {e}")
+        raise
+    finally:
+        await session.close()
+
+
+async def execute_postgres_query(query: str, params: dict = None):
+    """
+    Execute a raw SQL query.
+    
+    Args:
+        query: SQL query string
+        params: Query parameters dict
+    """
+    if not _engine:
+        logger.warning("PostgreSQL engine not available, skipping query")
+        return None
+    
+    async with get_postgres_session() as session:
+        if session is None:
+            return None
+        result = await session.execute(query, params or {})
+        await session.commit()
+        return result

app/staff/controllers/router.py

@@ -1,69 +1,148 @@
 """
-Employee API router - FastAPI endpoints for employee operations.
+Staff API router - FastAPI endpoints for staff operations.
+Simplified POS staff management.
 """
-from typing import Optional, List
-from fastapi import APIRouter, HTTPException, Query, Header, status
-# from insightfy_utils.logging import get_logger  # TODO: Uncomment when package is available
+from typing import Optional
+from fastapi import APIRouter, HTTPException, Query, status
 import logging
 
-from app.constants.employee_types import Designation, stafftatus
-from app.staff.schemas.schema import EmployeeCreate, EmployeeUpdate, EmployeeResponse
-from app.staff.services.service import staffervice
+from app.staff.schemas.staff_schema import (
+    StaffCreateSchema,
+    StaffUpdateSchema,
+    StaffResponseSchema,
+    StaffListResponse
+)
+from app.staff.services.staff_service import StaffService
 
-# logger = get_logger(__name__)  # TODO: Uncomment when insightfy_utils is available
 logger = logging.getLogger(__name__)
 
 router = APIRouter(
     prefix="/staff",
-    tags=["staff"],
+    tags=["Staff"],
     responses={404: {"description": "Not found"}},
 )
 
 
 @router.post(
     "",
-    response_model=EmployeeResponse,
+    response_model=StaffResponseSchema,
     status_code=status.HTTP_201_CREATED,
-    summary="Create a new employee",
-    description="""
-    Create a new employee with comprehensive validation:
-    - Validates employee code uniqueness
-    - Ensures email and phone uniqueness among active staff
-    - Enforces manager hierarchy rules
-    - Validates age requirements (minimum 18 years)
-    - Validates DOB/DOJ consistency
-    - Enforces 2FA for Admin/Finance/HR roles
-    - Validates location tracking consent requirements
-    """
+    summary="Create a new staff member"
 )
-async def create_employee(payload: EmployeeCreate) -> EmployeeResponse:
+async def create_staff(payload: StaffCreateSchema) -> StaffResponseSchema:
     """
-    Create a new employee.
+    Create a new staff member.
     
-    **Business Rules:**
-    - Employee code must be unique across all staff
-    - Email must be unique among active staff
-    - Phone must be unique among active staff
-    - Employee must be at least 18 years old
-    - DOJ cannot be more than 30 days in the future
-    - ASM must have an RSM manager
-    - BDE/Trainer must have ASM or RSM manager
-    - RSM/ASM must have a region assigned
-    - Admin/Finance/HR require 2FA enabled
-    - Location tracking requires mobile app access
+    **Required fields:**
+    - merchant_id: Store/merchant identifier
+    - name: Full name
+    - role: Position (e.g., Stylist, Cashier, Manager)
+    - contact: Phone and email
     
-    **Manager Hierarchy:**
-    - ASM → RSM
-    - BDE → ASM or RSM
-    - Trainer → ASM, RSM, or Head_Trainer
-    - Field_Sales → ASM, RSM, or BDE
+    **Optional fields:**
+    - skills: List of skills/specializations
+    - status: active (default), inactive, on_leave
+    - working_hours: Weekly schedule
+    - photo_url: Profile photo URL (HTTPS only)
+    - notes: Additional information
+    """
+    return await StaffService.create_staff(payload)
+
+
+@router.get(
+    "",
+    response_model=StaffListResponse,
+    summary="List staff members"
+)
+async def list_staff(
+    merchant_id: Optional[str] = Query(None, description="Filter by merchant ID"),
+    status: Optional[str] = Query(None, description="Filter by status (active, inactive, on_leave)"),
+    role: Optional[str] = Query(None, description="Filter by role"),
+    skip: int = Query(0, ge=0, description="Number of records to skip"),
+    limit: int = Query(100, ge=1, le=1000, description="Maximum records to return")
+):
+    """
+    List staff members with optional filters and pagination.
+    """
+    staff_list, total = await StaffService.list_staff(
+        merchant_id=merchant_id,
+        status=status,
+        role=role,
+        skip=skip,
+        limit=limit
+    )
+    
+    return StaffListResponse(
+        staff=staff_list,
+        total=total,
+        skip=skip,
+        limit=limit
+    )
+
+
+@router.get(
+    "/{staff_id}",
+    response_model=StaffResponseSchema,
+    summary="Get staff member by ID"
+)
+async def get_staff(staff_id: str) -> StaffResponseSchema:
+    """
+    Get detailed information about a specific staff member.
+    """
+    staff = await StaffService.get_staff_by_id(staff_id)
+    if not staff:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Staff {staff_id} not found"
+        )
+    return staff
+
+
+@router.put(
+    "/{staff_id}",
+    response_model=StaffResponseSchema,
+    summary="Update staff member"
+)
+async def update_staff(staff_id: str, payload: StaffUpdateSchema) -> StaffResponseSchema:
+    """
+    Update staff member information.
     
-    **Location Tracking:**
-    - Requires explicit consent with timestamp
-    - Background tracking requires location_tracking_consent
-    - Requires mobile app access (has_mobile_app=True)
+    All fields are optional - only provided fields will be updated.
     """
-    return await staffervice.create_employee(payload)
+    return await StaffService.update_staff(staff_id, payload)
+
+
+@router.delete(
+    "/{staff_id}",
+    summary="Delete staff member"
+)
+async def delete_staff(staff_id: str):
+    """
+    Delete a staff member (soft delete - sets status to inactive).
+    """
+    return await StaffService.delete_staff(staff_id)
+
+
+@router.get(
+    "/{staff_id}/schedule",
+    summary="Get staff schedule"
+)
+async def get_staff_schedule(staff_id: str):
+    """
+    Get working schedule for a staff member.
+    """
+    staff = await StaffService.get_staff_by_id(staff_id)
+    if not staff:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Staff {staff_id} not found"
+        )
+    
+    return {
+        "staff_id": staff.staff_id,
+        "name": staff.name,
+        "working_hours": staff.working_hours
+    }
 
 
 @router.get(

app/staff/models/staff_model.py

@@ -0,0 +1,75 @@
+"""
+Simplified Staff model for POS microservice.
+"""
+from datetime import datetime
+from typing import Optional, List, Dict, Any
+from pydantic import BaseModel, Field, EmailStr
+
+
+class ContactInfo(BaseModel):
+    """Staff contact information."""
+    phone: str = Field(..., description="Phone number with country code")
+    email: EmailStr = Field(..., description="Email address")
+
+
+class WorkingHours(BaseModel):
+    """Working hours for a specific day."""
+    day: str = Field(..., description="Day of week (Mon, Tue, Wed, Thu, Fri, Sat, Sun)")
+    from_time: str = Field(..., alias="from", description="Start time (HH:MM format)")
+    to_time: str = Field(..., alias="to", description="End time (HH:MM format)")
+    
+    class Config:
+        populate_by_name = True
+
+
+class StaffModel(BaseModel):
+    """
+    POS Staff data model.
+    Simplified model for salon/retail staff management.
+    """
+    staff_id: str = Field(..., description="Unique staff identifier")
+    merchant_id: str = Field(..., description="Merchant/store identifier")
+    
+    name: str = Field(..., min_length=1, max_length=100, description="Staff member name")
+    role: str = Field(..., description="Role/position (e.g., Stylist, Cashier, Manager)")
+    
+    contact: ContactInfo = Field(..., description="Contact information")
+    
+    skills: List[str] = Field(default_factory=list, description="List of skills/specializations")
+    status: str = Field(default="active", description="Status: active, inactive, on_leave")
+    
+    working_hours: List[WorkingHours] = Field(default_factory=list, description="Weekly working schedule")
+    
+    created_at: datetime = Field(default_factory=datetime.utcnow, description="Creation timestamp")
+    updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update timestamp")
+    
+    # Optional fields
+    photo_url: Optional[str] = Field(None, description="Profile photo URL")
+    notes: Optional[str] = Field(None, description="Additional notes")
+    metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "staff_id": "staff_01HZQX5K3N2P8R6T4V9W",
+                "merchant_id": "merchant_789xyz",
+                "name": "Aarav Sharma",
+                "role": "Stylist",
+                "contact": {
+                    "phone": "+91-9876543210",
+                    "email": "aarav@salon.com"
+                },
+                "skills": ["Haircut", "Color", "Styling"],
+                "status": "active",
+                "working_hours": [
+                    {"day": "Mon", "from": "10:00", "to": "19:00"},
+                    {"day": "Tue", "from": "10:00", "to": "19:00"},
+                    {"day": "Wed", "from": "10:00", "to": "19:00"},
+                    {"day": "Thu", "from": "10:00", "to": "19:00"},
+                    {"day": "Fri", "from": "10:00", "to": "19:00"},
+                    {"day": "Sat", "from": "11:00", "to": "18:00"}
+                ],
+                "created_at": "2024-01-15T10:30:00Z",
+                "updated_at": "2024-01-15T10:30:00Z"
+            }
+        }

app/staff/schemas/staff_schema.py

@@ -0,0 +1,101 @@
+"""
+Pydantic schemas for staff API requests and responses.
+"""
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from pydantic import BaseModel, Field, EmailStr, field_validator
+import re
+
+
+class ContactInfoSchema(BaseModel):
+    """Contact information schema."""
+    phone: str = Field(..., description="Phone number with country code", min_length=10, max_length=20)
+    email: EmailStr = Field(..., description="Email address")
+    
+    @field_validator('phone')
+    @classmethod
+    def validate_phone(cls, v):
+        """Validate phone number format."""
+        # Remove spaces and dashes for validation
+        phone = re.sub(r'[\s\-]', '', v)
+        if not re.match(r'^\+?[1-9]\d{7,14}$', phone):
+            raise ValueError('Invalid phone number format. Use international format with country code.')
+        return v
+
+
+class WorkingHoursSchema(BaseModel):
+    """Working hours schema."""
+    day: str = Field(..., description="Day of week", pattern="^(Mon|Tue|Wed|Thu|Fri|Sat|Sun)$")
+    from_time: str = Field(..., alias="from", description="Start time (HH:MM)", pattern="^([0-1][0-9]|2[0-3]):[0-5][0-9]$")
+    to_time: str = Field(..., alias="to", description="End time (HH:MM)", pattern="^([0-1][0-9]|2[0-3]):[0-5][0-9]$")
+    
+    class Config:
+        populate_by_name = True
+
+
+class StaffCreateSchema(BaseModel):
+    """Schema for creating a new staff member."""
+    merchant_id: str = Field(..., description="Merchant/store identifier", min_length=3, max_length=50)
+    name: str = Field(..., description="Staff member name", min_length=1, max_length=100)
+    role: str = Field(..., description="Role/position", min_length=1, max_length=50)
+    contact: ContactInfoSchema = Field(..., description="Contact information")
+    skills: List[str] = Field(default_factory=list, description="List of skills")
+    status: str = Field(default="active", description="Status", pattern="^(active|inactive|on_leave)$")
+    working_hours: List[WorkingHoursSchema] = Field(default_factory=list, description="Weekly schedule")
+    photo_url: Optional[str] = Field(None, description="Profile photo URL")
+    notes: Optional[str] = Field(None, description="Additional notes", max_length=500)
+    
+    @field_validator('photo_url')
+    @classmethod
+    def validate_photo_url(cls, v):
+        """Validate photo URL is HTTPS."""
+        if v and not v.startswith('https://'):
+            raise ValueError('Photo URL must use HTTPS')
+        return v
+
+
+class StaffUpdateSchema(BaseModel):
+    """Schema for updating staff information."""
+    name: Optional[str] = Field(None, description="Staff member name", min_length=1, max_length=100)
+    role: Optional[str] = Field(None, description="Role/position", min_length=1, max_length=50)
+    contact: Optional[ContactInfoSchema] = Field(None, description="Contact information")
+    skills: Optional[List[str]] = Field(None, description="List of skills")
+    status: Optional[str] = Field(None, description="Status", pattern="^(active|inactive|on_leave)$")
+    working_hours: Optional[List[WorkingHoursSchema]] = Field(None, description="Weekly schedule")
+    photo_url: Optional[str] = Field(None, description="Profile photo URL")
+    notes: Optional[str] = Field(None, description="Additional notes", max_length=500)
+    
+    @field_validator('photo_url')
+    @classmethod
+    def validate_photo_url(cls, v):
+        """Validate photo URL is HTTPS."""
+        if v and not v.startswith('https://'):
+            raise ValueError('Photo URL must use HTTPS')
+        return v
+
+
+class StaffResponseSchema(BaseModel):
+    """Schema for staff API responses."""
+    staff_id: str
+    merchant_id: str
+    name: str
+    role: str
+    contact: ContactInfoSchema
+    skills: List[str]
+    status: str
+    working_hours: List[WorkingHoursSchema]
+    photo_url: Optional[str] = None
+    notes: Optional[str] = None
+    created_at: datetime
+    updated_at: datetime
+    
+    class Config:
+        from_attributes = True
+
+
+class StaffListResponse(BaseModel):
+    """Schema for staff list response."""
+    staff: List[StaffResponseSchema]
+    total: int
+    skip: int
+    limit: int

app/staff/services/staff_service.py

@@ -0,0 +1,276 @@
+"""
+Staff service layer - business logic and database operations.
+Syncs staff data to both MongoDB and PostgreSQL (trans.pos_staff_ref).
+"""
+from datetime import datetime
+from typing import Optional, List, Dict, Any
+from fastapi import HTTPException, status
+import logging
+import secrets
+from sqlalchemy import text
+
+from app.nosql import get_database
+from app.sql import get_postgres_session
+from app.constants.collections import POS_STAFF_COLLECTION
+from app.staff.models.staff_model import StaffModel
+from app.staff.schemas.staff_schema import StaffCreateSchema, StaffUpdateSchema, StaffResponseSchema
+
+logger = logging.getLogger(__name__)
+
+
+
+
+async def sync_staff_to_postgres(staff_id: str, merchant_id: str, staff_name: str):
+    """
+    Sync staff data to PostgreSQL trans.pos_staff_ref table.
+    
+    Args:
+        staff_id: Staff UUID
+        merchant_id: Merchant UUID  
+        staff_name: Staff member name
+    """
+    try:
+        async with get_postgres_session() as session:
+            if session is None:
+                logger.warning("PostgreSQL not available, skipping staff sync")
+                return
+            
+            query = text("""
+                INSERT INTO trans.pos_staff_ref (staff_id, merchant_id, staff_name, created_at, updated_at)
+                VALUES (:staff_id, :merchant_id, :staff_name, NOW(), NOW())
+                ON CONFLICT (staff_id) 
+                DO UPDATE SET 
+                    staff_name = EXCLUDED.staff_name,
+                    updated_at = NOW()
+            """)
+            
+            await session.execute(query, {
+                "staff_id": staff_id,
+                "merchant_id": merchant_id,
+                "staff_name": staff_name
+            })
+            await session.commit()
+            logger.info(f"Synced staff {staff_id} to trans.pos_staff_ref")
+            
+    except Exception as e:
+        logger.error(f"Failed to sync staff {staff_id} to PostgreSQL: {e}")
+        # Don't raise - PostgreSQL sync is secondary to MongoDB
+def generate_staff_id() -> str:
+    """Generate a unique staff ID."""
+    return f"staff_{secrets.token_urlsafe(16)}"
+
+
+class StaffService:
+    """Service class for staff operations."""
+
+    @staticmethod
+    async def create_staff(payload: StaffCreateSchema) -> StaffResponseSchema:
+        """
+        Create a new staff member.
+        
+        Args:
+            payload: Staff creation data
+            
+        Returns:
+            Created staff response
+        """
+        try:
+            # Generate staff ID
+            staff_id = generate_staff_id()
+            
+            # Create staff model
+            now = datetime.utcnow()
+            staff_data = payload.model_dump(by_alias=True)
+            staff_data["staff_id"] = staff_id
+            staff_data["created_at"] = now
+            staff_data["updated_at"] = now
+            
+            # Insert into database
+            await get_database()[POS_STAFF_COLLECTION].insert_one(staff_data)
+            
+            # Sync to PostgreSQL trans.pos_staff_ref
+            await sync_staff_to_postgres(
+                staff_id=staff_id,
+                merchant_id=payload.merchant_id,
+                staff_name=payload.name
+            )
+            
+            logger.info(f"Created staff {staff_id} for merchant {payload.merchant_id}")
+            
+            # Return response
+            return StaffResponseSchema(**staff_data)
+            
+        except Exception as e:
+            logger.error(f"Error creating staff: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail=f"Error creating staff: {str(e)}"
+            )
+
+    @staticmethod
+    async def get_staff_by_id(staff_id: str) -> Optional[StaffResponseSchema]:
+        """Get staff by ID."""
+        try:
+            staff = await get_database()[POS_STAFF_COLLECTION].find_one({"staff_id": staff_id})
+            if not staff:
+                return None
+            return StaffResponseSchema(**staff)
+        except Exception as e:
+            logger.error(f"Error fetching staff {staff_id}: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Error retrieving staff"
+            )
+
+    @staticmethod
+    async def update_staff(staff_id: str, payload: StaffUpdateSchema) -> StaffResponseSchema:
+        """
+        Update staff information.
+        
+        Args:
+            staff_id: Staff ID to update
+            payload: Update data
+            
+        Returns:
+            Updated staff response
+        """
+        # Check staff exists
+        existing = await StaffService.get_staff_by_id(staff_id)
+        if not existing:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"Staff {staff_id} not found"
+            )
+
+        # Prepare update data
+        update_data = payload.model_dump(exclude_unset=True, by_alias=True)
+        if not update_data:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="No update data provided"
+            )
+
+        # Add updated timestamp
+        update_data["updated_at"] = datetime.utcnow()
+
+        try:
+            # Update in database
+            result = await get_database()[POS_STAFF_COLLECTION].update_one(
+                {"staff_id": staff_id},
+                {"$set": update_data}
+            )
+# Sync to PostgreSQL if name was updated
+            if "name" in update_data:
+                await sync_staff_to_postgres(
+                    staff_id=staff_id,
+                    merchant_id=updated_staff.merchant_id,
+                    staff_name=updated_staff.name
+                )
+            
+            
+            if result.modified_count == 0:
+                logger.warning(f"No changes made to staff {staff_id}")
+
+            # Fetch updated staff
+            updated_staff = await StaffService.get_staff_by_id(staff_id)
+            
+            logger.info(f"Updated staff {staff_id}")
+            
+            return updated_staff
+            
+        except Exception as e:
+            logger.error(f"Error updating staff {staff_id}: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail=f"Error updating staff: {str(e)}"
+            )
+
+    @staticmethod
+    async def list_staff(
+        merchant_id: Optional[str] = None,
+        status: Optional[str] = None,
+        role: Optional[str] = None,
+        skip: int = 0,
+        limit: int = 100
+    ) -> tuple[List[StaffResponseSchema], int]:
+        """
+        List staff with optional filters.
+        
+        Args:
+            merchant_id: Filter by merchant
+            status: Filter by status
+            role: Filter by role
+            skip: Number of records to skip
+            limit: Maximum records to return
+            
+        Returns:
+            Tuple of (staff list, total count)
+        """
+        try:
+            # Build query
+            query = {}
+            if merchant_id:
+                query["merchant_id"] = merchant_id
+            if status:
+                query["status"] = status
+            if role:
+                query["role"] = role
+
+            # Get total count
+            total = await get_database()[POS_STAFF_COLLECTION].count_documents(query)
+
+            # Fetch staff
+            cursor = get_database()[POS_STAFF_COLLECTION].find(query).skip(skip).limit(limit)
+            staff_list = await cursor.to_list(length=limit)
+            
+            staff_responses = [StaffResponseSchema(**staff) for staff in staff_list]
+            
+            return staff_responses, total
+            
+        except Exception as e:
+            logger.error(f"Error listing staff: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Error listing staff"
+            )
+
+    @staticmethod
+    async def delete_staff(staff_id: str) -> Dict[str, str]:
+        """
+        Delete a staff member (soft delete by setting status to inactive).
+        
+        Args:
+            staff_id: Staff ID to delete
+            
+        Returns:
+            Success message
+        """
+        # Check staff exists
+        existing = await StaffService.get_staff_by_id(staff_id)
+        if not existing:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"Staff {staff_id} not found"
+            )
+
+        try:
+            # Soft delete - set status to inactive
+            await get_database()[POS_STAFF_COLLECTION].update_one(
+                {"staff_id": staff_id},
+                {
+                    "$set": {
+                        "status": "inactive",
+                        "updated_at": datetime.utcnow()
+                    }
+                }
+            )
+            
+            logger.info(f"Deleted staff {staff_id}")
+            return {"message": f"Staff {staff_id} deleted successfully"}
+            
+        except Exception as e:
+            logger.error(f"Error deleting staff {staff_id}: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Error deleting staff"
+            )

requirements.txt

@@ -4,13 +4,15 @@ python-multipart==0.0.6
 
 motor==3.3.2
 pymongo==4.6.0
+sqlalchemy[asyncio]>=2.0.36
+psycopg[binary]==3.3.2
 email-validator==2.3.0
 redis==5.0.1
 # insightfy-utils>=0.1.0  # TODO: Add back when package is available
 
 python-jose[cryptography]==3.3.0
 passlib[bcrypt]==1.7.4
-bcrypt==4.1.3
+bcrypt==4.0.1
 pydantic>=2.12.5,<3.0.0
 pydantic-settings>=2.0.0