Upload 32 files

app/core/auth.py

@@ -1,25 +1,17 @@
-from datetime import datetime
-from fastapi import HTTPException, Security, status
-from fastapi.security import APIKeyHeader
-from ..core.config import settings
-
-# API Key header scheme for service token
-SERVICE_TOKEN_HEADER = APIKeyHeader(name="X-Service-Token", auto_error=False)
-
-async def verify_service_token(token: str = Security(SERVICE_TOKEN_HEADER)) -> bool:
-    """Verify the service token for internal service-to-service communication."""
-    if not token:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Service token is required",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-    
-    if token != settings.SERVICE_TOKEN:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid service token",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-    
-    return True
+from fastapi import Depends, HTTPException, status, Header
+from typing import Optional
+from .config import settings
+
+async def verify_service_token(
+    x_service_token: Optional[str] = Header(None, alias="X-Service-Token")
+) -> None:
+    if not x_service_token:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="X-Service-Token header is missing",
+        )
+    if x_service_token != settings.SERVICE_TOKEN:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid service token",
+        )

app/core/config.py

@@ -1,17 +1,40 @@
-from pydantic_settings import BaseSettings
-
-class Settings(BaseSettings):
-    PROJECT_NAME: str = "POS Backend"
-    API_V1_STR: str = "/api/v1"
-    SECRET_KEY: str = "your-secret-key-here"
-    ALGORITHM: str = "HS256"
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
-    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
-    DATABASE_URL: str = "postgresql+asyncpg://postgres.juycnkjuzylnbruwaqmp:Lovyelias5584.@aws-0-eu-central-1.pooler.supabase.com:5432/postgres"
-    SERVICE_TOKEN: str = "6f9aee88d5b94b0190c317abcbf4e6e7834cc2c7f02e45693e123456789abcde"  # Should be set in .env file
-
-
-    class Config:
-        env_file = ".env"
-
-settings = Settings()
+from pydantic_settings import BaseSettings
+from typing import ClassVar
+
+
+class Settings(BaseSettings):
+    API_V1_STR: str = "/api/v1"
+    PROJECT_NAME: str = "Admin Dashboard"
+    VERSION: str = "1.0.0"
+
+    # Service Authentication
+    SERVICE_TOKEN: str = "6f9aee88d5b94b0190c317abcbf4e6e7834cc2c7f02e45693e123456789abcde"
+    POS_API_URL: str = "https://fred808-desk-back2.hf.space"  # Default value, should be overridden in environment
+
+    # PostgreSQL Database settings
+    DATABASE_URL: ClassVar[str] = "postgresql+asyncpg://postgres.juycnkjuzylnbruwaqmp:Lovyelias5584.@aws-0-eu-central-1.pooler.supabase.com:5432/postgres"
+
+    # JWT Settings
+    SECRET_KEY: str = "your-secret-key-here"
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+
+    # Redis settings
+    REDIS_HOST: str = "localhost"
+    REDIS_PORT: int = 6379
+
+    # Email settings
+    MAIL_USERNAME: str = "yungdml31@gmail.com"
+    MAIL_PASSWORD: str = ""
+    MAIL_FROM: str = "admin@angelo.com"
+    MAIL_PORT: int = 587
+    MAIL_SERVER: str = "smtp.gmail.com"
+
+    # Frontend URL
+    FRONTEND_URL: str = "http://localhost:3000"
+
+    class Config:
+        case_sensitive = True
+
+
+settings = Settings()

app/core/dependencies.py

@@ -0,0 +1,53 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+from jose import JWTError, jwt
+from ..db.database import get_db
+from ..db.models import User
+from ..core.config import settings
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
+
+async def get_current_user(
+    token: str = Depends(oauth2_scheme),
+    db: AsyncSession = Depends(get_db)
+):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    
+    try:
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+        user_id: str = payload.get("sub")
+        if user_id is None:
+            raise credentials_exception
+    except JWTError:
+        raise credentials_exception
+    
+    stmt = select(User).options(selectinload(User.roles)).where(User.id == int(user_id))
+    result = await db.execute(stmt)
+    user = result.scalar_one_or_none()
+    
+    if user is None:
+        raise credentials_exception
+    return user
+
+async def get_current_active_user(
+    current_user: User = Depends(get_current_user)
+):
+    if not current_user.is_active:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    return current_user
+
+async def get_current_superuser(
+    current_user: User = Depends(get_current_user)
+):
+    if not current_user.is_superuser:
+        raise HTTPException(
+            status_code=403, detail="The user doesn't have enough privileges"
+        )
+    return current_user

app/core/security.py

@@ -0,0 +1,23 @@
+from datetime import datetime, timedelta
+from typing import Any, Optional
+from jose import jwt
+from passlib.context import CryptContext
+from .config import settings
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+def create_access_token(subject: Any, expires_delta: Optional[timedelta] = None) -> str:
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    
+    to_encode = {"exp": expire, "sub": str(subject)}
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return encoded_jwt
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password: str) -> str:
+    return pwd_context.hash(password)

app/db/database.py

@@ -1,83 +1,55 @@
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
-from sqlalchemy.orm import DeclarativeBase
-from sqlalchemy import text
+from sqlalchemy.orm import declarative_base
 from ..core.config import settings
-import logging
-import ssl
-import platform
-import asyncio
+import contextlib
 
-logger = logging.getLogger(__name__)
-
-def get_ssl_context():
-    """Create an SSL context for database connection."""
-    ssl_context = ssl.create_default_context()
-    ssl_context.check_hostname = False
-    ssl_context.verify_mode = ssl.CERT_NONE
-    return ssl_context
-
-# Configure Windows-specific event loop policy
-if platform.system() == 'Windows':
-    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
-
-# Create async engine with connection pooling
-engine = create_async_engine(
+# Create async engine for FastAPI
+async_engine = create_async_engine(
     settings.DATABASE_URL,
     echo=True,
-    pool_pre_ping=True,
-    pool_size=20,
-    max_overflow=10,
-    pool_timeout=30,
-    connect_args={
-        "ssl": get_ssl_context(),
-        "server_settings": {
-            "application_name": "pos_backend",
-            "statement_timeout": "60000",  # 60 seconds
-        }
-    }
+    future=True,
+    pool_pre_ping=True
 )
 
-# Create async session maker with pooling configuration
-async_session_maker = async_sessionmaker(
-    engine,
+# Create async session factory
+AsyncSessionLocal = async_sessionmaker(
+    bind=async_engine,
     class_=AsyncSession,
-    expire_on_commit=False,
-    autocommit=False,
-    autoflush=False
+    expire_on_commit=False
 )
 
-class Base(DeclarativeBase):
-    pass
-
-async def verify_connection():
-    try:
-        async with engine.connect() as conn:
-            await conn.execute(text("SELECT 1"))
-            logger.info("Database connection verified successfully")
-            return True
-    except Exception as e:
-        logger.error(f"Database connection failed: {str(e)}")
-        raise
+# Create declarative base for models
+Base = declarative_base()
 
+# Database dependency for FastAPI routes
 async def get_db():
-    async with async_session_maker() as session:
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+        finally:
+            await session.close()
+
+# Database access for background tasks and services
+class Database:
+    def __init__(self):
+        self._session_factory = AsyncSessionLocal
+
+    @contextlib.asynccontextmanager
+    async def session(self):
+        """Get a database session with automatic commit/rollback"""
+        session = self._session_factory()
         try:
-            await verify_connection()
             yield session
             await session.commit()
-        except Exception as e:
-            logger.error(f"Database session error: {str(e)}")
+        except:
             await session.rollback()
             raise
         finally:
             await session.close()
 
-async def init_db():
-    try:
-        await verify_connection()
-        async with engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-        logger.info("Database initialized successfully")
-    except Exception as e:
-        logger.error(f"Database initialization failed: {str(e)}")
-        raise
+    async def get_session(self):
+        """Get a session for manual management"""
+        return self._session_factory()
+
+# Create singleton instance for database access
+db = Database()

app/db/init_db.py

@@ -0,0 +1,77 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from ..core.config import settings
+from ..core.security import get_password_hash
+from datetime import datetime
+from .models import Base, User, Product
+import asyncio
+
+def init_db():
+    # Create synchronous engine for initialization
+    engine = create_engine(
+        settings.DATABASE_URL.replace("+asyncpg", ""),
+        echo=True
+    )
+    
+    # Create all tables
+    Base.metadata.create_all(bind=engine)
+    
+    # Create session
+    SessionLocal = sessionmaker(bind=engine)
+    session = SessionLocal()
+    
+    try:
+        # Create default admin user if not exists
+        admin_user = session.query(User).filter_by(email="admin@example.com").first()
+        if not admin_user:
+            admin_user = User(
+                email="admin@example.com",
+                username="admin",
+                full_name="System Administrator",
+                hashed_password=get_password_hash("admin123"),  # Change in production
+                is_active=True,
+                is_superuser=True,
+                roles=["admin"],
+                created_at=datetime.utcnow()
+            )
+            session.add(admin_user)
+            print("Created default admin user.")
+        
+        # Create default product categories as products
+        categories = [
+            "Soups & Stews", 
+            "Rice Dishes", 
+            "Swallow & Fufu", 
+            "Snacks & Small Chops", 
+            "Protein & Meat", 
+            "Drinks"
+        ]
+        
+        for category in categories:
+            exists = session.query(Product).filter_by(name=category).first()
+            if not exists:
+                product = Product(
+                    name=category,
+                    description=f"Category: {category}",
+                    price=0.0,  # Category products have zero price
+                    category=category,
+                    inventory_count=0,  # Categories don't have inventory
+                    seller_id=admin_user.id if admin_user else 1,  # Link to admin user
+                    created_at=datetime.utcnow()
+                )
+                session.add(product)
+        
+        print("Initialized product categories.")
+        
+        # Commit changes
+        session.commit()
+        
+    except Exception as e:
+        print(f"Error during initialization: {e}")
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+if __name__ == "__main__":
+    init_db()

app/db/models.py

@@ -0,0 +1,186 @@
+from sqlalchemy import Column, Integer, String, Boolean, DateTime, Float, ForeignKey, ARRAY, JSON, Table
+from sqlalchemy.orm import relationship, mapped_column, Mapped
+from sqlalchemy.dialects.postgresql import JSONB
+from datetime import datetime
+from typing import List, Optional
+from .database import Base
+
+# Association tables for many-to-many relationships
+user_roles = Table(
+    'user_roles',
+    Base.metadata,
+    Column('user_id', Integer, ForeignKey('users.id')),
+    Column('role_id', Integer, ForeignKey('roles.id'))
+)
+
+# Role model
+class Role(Base):
+    __tablename__ = "roles"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, unique=True, index=True)
+    description: Mapped[str] = mapped_column(String)
+    permissions: Mapped[List[str]] = mapped_column(ARRAY(String), default=list)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+
+    # Relationships
+    users = relationship("User", secondary=user_roles, back_populates="roles")
+
+# Branch model
+class Branch(Base):
+    __tablename__ = "branches"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, unique=True, index=True)
+    address: Mapped[str] = mapped_column(String)
+    phone: Mapped[str] = mapped_column(String)
+    email: Mapped[str] = mapped_column(String)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+
+    # Relationships
+    users = relationship("User", back_populates="branch")
+    products = relationship("Product", back_populates="branch")
+    orders = relationship("Order", back_populates="branch")
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    email: Mapped[str] = mapped_column(String, unique=True, index=True)
+    username: Mapped[str] = mapped_column(String, unique=True, index=True)
+    full_name: Mapped[str] = mapped_column(String)
+    hashed_password: Mapped[str] = mapped_column(String)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    is_superuser: Mapped[bool] = mapped_column(Boolean, default=False)
+    branch_id: Mapped[Optional[int]] = mapped_column(ForeignKey("branches.id"))
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+
+    # Update roles relationship to use many-to-many
+    roles = relationship("Role", secondary=user_roles, back_populates="users")
+    
+    # Other relationships
+    branch = relationship("Branch", back_populates="users")
+    products = relationship("Product", back_populates="seller")
+    orders = relationship("Order", back_populates="customer")
+    notifications = relationship("Notification", back_populates="user")
+    sessions = relationship("Session", back_populates="user", cascade="all, delete-orphan")
+
+class Product(Base):
+    __tablename__ = "products"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, index=True)
+    description: Mapped[str] = mapped_column(String)
+    price: Mapped[float] = mapped_column(Float)
+    category: Mapped[str] = mapped_column(String, index=True)
+    inventory_count: Mapped[int] = mapped_column(Integer)
+    seller_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
+    branch_id: Mapped[int] = mapped_column(ForeignKey("branches.id"))
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, 
+        default=datetime.utcnow, 
+        onupdate=datetime.utcnow
+    )
+
+    # Relationships
+    seller = relationship("User", back_populates="products")
+    branch = relationship("Branch", back_populates="products")
+    order_items = relationship("OrderItem", back_populates="product")
+
+class Order(Base):
+    __tablename__ = "orders"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    customer_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
+    branch_id: Mapped[int] = mapped_column(ForeignKey("branches.id"))
+    total_amount: Mapped[float] = mapped_column(Float)
+    status: Mapped[str] = mapped_column(String)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, 
+        default=datetime.utcnow, 
+        onupdate=datetime.utcnow
+    )
+
+    # Relationships
+    customer = relationship("User", back_populates="orders")
+    branch = relationship("Branch", back_populates="orders")
+    items = relationship("OrderItem", back_populates="order", cascade="all, delete-orphan")
+
+class OrderItem(Base):
+    __tablename__ = "order_items"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"))
+    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"))
+    quantity: Mapped[int] = mapped_column(Integer)
+    price: Mapped[float] = mapped_column(Float)
+
+    # Relationships
+    order = relationship("Order", back_populates="items")
+    product = relationship("Product", back_populates="order_items")
+
+class Notification(Base):
+    __tablename__ = "notifications"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
+    title: Mapped[str] = mapped_column(String)
+    message: Mapped[str] = mapped_column(String)
+    type: Mapped[str] = mapped_column(String)
+    data: Mapped[Optional[dict]] = mapped_column(JSONB)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    read: Mapped[bool] = mapped_column(Boolean, default=False)
+
+    # Relationship
+    user = relationship("User", back_populates="notifications")
+
+class Event(Base):
+    __tablename__ = "events"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
+    title: Mapped[str] = mapped_column(String)
+    description: Mapped[str] = mapped_column(String)
+    start_time: Mapped[datetime] = mapped_column(DateTime)
+    end_time: Mapped[datetime] = mapped_column(DateTime)
+    attendees: Mapped[List[str]] = mapped_column(ARRAY(String), default=list)
+    is_all_day: Mapped[bool] = mapped_column(Boolean, default=False)
+    reminder_minutes: Mapped[int] = mapped_column(Integer)
+    status: Mapped[str] = mapped_column(String)
+    attendee_responses: Mapped[dict] = mapped_column(JSONB, default=dict)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, 
+        default=datetime.utcnow, 
+        onupdate=datetime.utcnow
+    )
+    # Fields for recurring events
+    is_recurring: Mapped[bool] = mapped_column(Boolean, default=False)
+    recurrence_pattern: Mapped[Optional[str]] = mapped_column(String)
+    recurrence_group: Mapped[Optional[str]] = mapped_column(String)
+    recurrence_end_date: Mapped[Optional[datetime]] = mapped_column(DateTime)
+    parent_event_id: Mapped[Optional[int]] = mapped_column(Integer)
+    sequence_number: Mapped[Optional[int]] = mapped_column(Integer)
+    reminder_sent: Mapped[bool] = mapped_column(Boolean, default=False)
+
+    # Relationship
+    user = relationship("User")
+
+class Session(Base):
+    __tablename__ = "sessions"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
+    session_token: Mapped[str] = mapped_column(String, unique=True, index=True)
+    last_activity: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    expires_at: Mapped[datetime] = mapped_column(DateTime)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+
+    # Relationship
+    user = relationship("User", back_populates="sessions")

app/db/schemas.py

@@ -0,0 +1,318 @@
+from sqlalchemy.orm import validates
+from sqlalchemy import event
+from datetime import datetime
+from typing import List, Optional, Dict
+from pydantic import BaseModel, EmailStr, validator
+from .models import User, Product, Order, Event, Notification, Session
+import re
+
+# Role schemas
+class RoleBase(BaseModel):
+    name: str
+    description: str
+    permissions: List[str] = []
+
+class RoleCreate(RoleBase):
+    pass
+
+class RoleUpdate(RoleBase):
+    name: Optional[str] = None
+    description: Optional[str] = None
+    permissions: Optional[List[str]] = None
+
+class RoleInDB(RoleBase):
+    id: int
+    created_at: datetime
+    updated_at: Optional[datetime] = None
+
+    class Config:
+        from_attributes = True
+
+class BranchBase(BaseModel):
+    name: str
+    address: str
+    phone: str
+    email: EmailStr
+    is_active: bool = True
+
+class BranchCreate(BranchBase):
+    pass
+
+class BranchInDB(BranchBase):
+    id: int
+    created_at: datetime
+    updated_at: Optional[datetime] = None
+
+    class Config:
+        from_attributes = True
+
+# Update User schemas
+class UserBase(BaseModel):
+    email: EmailStr
+    username: str
+    full_name: str
+    is_active: bool = True
+    is_superuser: bool = False
+    branch_id: Optional[int] = None
+
+class UserCreate(UserBase):
+    password: str
+    role_ids: Optional[List[int]] = None  # IDs of roles to assign
+
+class UserUpdate(BaseModel):
+    email: Optional[EmailStr] = None
+    username: Optional[str] = None
+    full_name: Optional[str] = None
+    is_active: Optional[bool] = None
+    is_superuser: Optional[bool] = None
+    password: Optional[str] = None
+    branch_id: Optional[int] = None
+    role_ids: Optional[List[int]] = None
+
+class UserInDB(UserBase):
+    id: int
+    created_at: datetime
+    roles: List[RoleInDB]
+
+    class Config:
+        from_attributes = True
+
+class ProductBase(BaseModel):
+    name: str
+    description: str
+    price: float
+    category: str
+    inventory_count: int
+    seller_id: int
+    branch_id: int
+
+class ProductCreate(ProductBase):
+    pass
+
+class ProductInDB(ProductBase):
+    id: int
+    created_at: datetime
+    updated_at: Optional[datetime] = None
+
+    class Config:
+        from_attributes = True
+
+class OrderItemBase(BaseModel):
+    product_id: int
+    quantity: int
+    price: float
+
+class OrderItemCreate(OrderItemBase):
+    pass
+
+class OrderItemInDB(OrderItemBase):
+    id: int
+    order_id: int
+
+    class Config:
+        from_attributes = True
+
+class OrderBase(BaseModel):
+    customer_id: int
+    branch_id: int
+    total_amount: float
+    status: str = "pending"
+    items: List[OrderItemCreate]
+
+class OrderCreate(OrderBase):
+    pass
+
+class OrderInDB(OrderBase):
+    id: int
+    created_at: datetime
+    updated_at: Optional[datetime] = None
+    items: List[OrderItemInDB]
+
+    class Config:
+        from_attributes = True
+
+class NotificationBase(BaseModel):
+    user_id: int
+    title: str
+    message: str
+    type: str
+    data: Optional[dict] = None
+    read: bool = False
+
+class NotificationCreate(NotificationBase):
+    pass
+
+class NotificationInDB(NotificationBase):
+    id: int
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
+
+class EventBase(BaseModel):
+    title: str
+    description: str
+    start_time: datetime
+    end_time: datetime
+    is_all_day: bool = False
+    reminder_minutes: int = 30
+
+    @validator('end_time')
+    def end_time_after_start_time(cls, v, values):
+        if 'start_time' in values and v <= values['start_time']:
+            raise ValueError('end_time must be after start_time')
+        return v
+
+    @validator('reminder_minutes')
+    def valid_reminder_minutes(cls, v):
+        if v < 0:
+            raise ValueError('reminder_minutes cannot be negative')
+        return v
+
+class EventCreate(EventBase):
+    attendees: List[str] = []
+
+class EventUpdate(BaseModel):
+    title: Optional[str] = None
+    description: Optional[str] = None
+    start_time: Optional[datetime] = None
+    end_time: Optional[datetime] = None
+    is_all_day: Optional[bool] = None
+    reminder_minutes: Optional[int] = None
+    attendees: Optional[List[str]] = None
+
+    @validator('reminder_minutes')
+    def valid_reminder_minutes(cls, v):
+        if v is not None and v < 0:
+            raise ValueError('reminder_minutes cannot be negative')
+        return v
+
+class EventInDB(EventBase):
+    id: int
+    user_id: int
+    attendees: List[str]
+    status: str
+    attendee_responses: Dict[str, str]
+    created_at: datetime
+    updated_at: Optional[datetime] = None
+    reminder_sent: bool = False
+    is_recurring: bool = False
+    recurrence_group: Optional[str] = None
+    parent_event_id: Optional[int] = None
+    sequence_number: Optional[int] = None
+
+    class Config:
+        from_attributes = True
+
+class RecurringEventCreate(EventCreate):
+    recurrence_pattern: str
+    recurrence_end_date: Optional[datetime] = None
+
+    @validator('recurrence_pattern')
+    def valid_recurrence_pattern(cls, v):
+        valid_patterns = ['daily', 'weekly', 'monthly', 'yearly']
+        if v not in valid_patterns:
+            raise ValueError(f'recurrence_pattern must be one of: {", ".join(valid_patterns)}')
+        return v
+
+    @validator('recurrence_end_date')
+    def end_date_after_start_time(cls, v, values):
+        if v is not None and 'start_time' in values and v <= values['start_time']:
+            raise ValueError('recurrence_end_date must be after start_time')
+        return v
+
+class RecurringEventUpdate(EventUpdate):
+    recurrence_pattern: Optional[str] = None
+    recurrence_end_date: Optional[datetime] = None
+
+    @validator('recurrence_pattern')
+    def valid_recurrence_pattern(cls, v):
+        if v is not None:
+            valid_patterns = ['daily', 'weekly', 'monthly', 'yearly']
+            if v not in valid_patterns:
+                raise ValueError(f'recurrence_pattern must be one of: {", ".join(valid_patterns)}')
+        return v
+
+class SessionBase(BaseModel):
+    user_id: int
+    session_token: str
+    expires_at: datetime
+    is_active: bool = True
+
+class SessionCreate(SessionBase):
+    pass
+
+class SessionUpdate(BaseModel):
+    last_activity: Optional[datetime] = None
+    expires_at: Optional[datetime] = None
+    is_active: Optional[bool] = None
+
+class SessionInDB(SessionBase):
+    id: int
+    created_at: datetime
+    last_activity: datetime
+
+    class Config:
+        from_attributes = True
+
+class LoginData(BaseModel):
+    email: str
+    password: str
+
+@validates('email')
+def validate_email(self, key, email):
+    if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email):
+        raise ValueError('Invalid email address')
+    return email
+
+@validates('username')
+def validate_username(self, key, username):
+    if len(username) < 3:
+        raise ValueError('Username must be at least 3 characters long')
+    return username
+
+@validates('inventory_count')
+def validate_inventory(self, key, count):
+    if count < 0:
+        raise ValueError('Inventory count cannot be negative')
+    return count
+
+@validates('price')
+def validate_price(self, key, price):
+    if price < 0:
+        raise ValueError('Price cannot be negative')
+    return price
+
+# Event listeners for automatic timestamps
+@event.listens_for(Product, 'before_insert')
+def set_created_at(mapper, connection, target):
+    target.created_at = datetime.utcnow()
+    target.updated_at = datetime.utcnow()
+
+@event.listens_for(Product, 'before_update')
+def set_updated_at(mapper, connection, target):
+    target.updated_at = datetime.utcnow()
+
+@event.listens_for(Order, 'before_insert')
+def set_order_created_at(mapper, connection, target):
+    target.created_at = datetime.utcnow()
+    target.updated_at = datetime.utcnow()
+
+@event.listens_for(Order, 'before_update')
+def set_order_updated_at(mapper, connection, target):
+    target.updated_at = datetime.utcnow()
+
+@event.listens_for(Event, 'before_insert')
+def set_event_created_at(mapper, connection, target):
+    target.created_at = datetime.utcnow()
+    target.updated_at = datetime.utcnow()
+
+@event.listens_for(Event, 'before_update')
+def set_event_updated_at(mapper, connection, target):
+    target.updated_at = datetime.utcnow()
+
+# Add validators to models
+User.validate_email = validate_email
+User.validate_username = validate_username
+Product.validate_inventory = validate_inventory
+Product.validate_price = validate_price

app/schemas/events.py

@@ -0,0 +1,87 @@
+from pydantic import BaseModel, validator
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+
+class EventBase(BaseModel):
+    title: str
+    description: str
+    start_time: datetime
+    end_time: datetime
+    is_all_day: bool = False
+    reminder_minutes: int = 30
+
+    @validator('end_time')
+    def end_time_after_start_time(cls, v, values):
+        if 'start_time' in values and v <= values['start_time']:
+            raise ValueError('end_time must be after start_time')
+        return v
+
+    @validator('reminder_minutes')
+    def valid_reminder_minutes(cls, v):
+        if v < 0:
+            raise ValueError('reminder_minutes cannot be negative')
+        return v
+
+class EventCreate(EventBase):
+    attendees: List[str] = []
+
+class EventUpdate(BaseModel):
+    title: Optional[str] = None
+    description: Optional[str] = None
+    start_time: Optional[datetime] = None
+    end_time: Optional[datetime] = None
+    is_all_day: Optional[bool] = None
+    reminder_minutes: Optional[int] = None
+    attendees: Optional[List[str]] = None
+
+    @validator('reminder_minutes')
+    def valid_reminder_minutes(cls, v):
+        if v is not None and v < 0:
+            raise ValueError('reminder_minutes cannot be negative')
+        return v
+
+class EventInDB(EventBase):
+    id: str
+    user_id: str
+    attendees: List[str]
+    status: str
+    attendee_responses: Dict[str, str]
+    created_at: datetime
+    updated_at: Optional[datetime] = None
+    reminder_sent: bool = False
+    is_recurring: bool = False
+    recurrence_group: Optional[str] = None
+    parent_event_id: Optional[str] = None
+    sequence_number: Optional[int] = None
+
+    class Config:
+        orm_mode = True
+
+class RecurringEventCreate(EventCreate):
+    recurrence_pattern: str
+    recurrence_end_date: Optional[datetime] = None
+
+    @validator('recurrence_pattern')
+    def valid_recurrence_pattern(cls, v):
+        valid_patterns = ['daily', 'weekly', 'monthly', 'yearly']
+        if v not in valid_patterns:
+            raise ValueError(f'recurrence_pattern must be one of: {", ".join(valid_patterns)}')
+        return v
+
+    @validator('recurrence_end_date')
+    def end_date_after_start_time(cls, v, values):
+        if v is not None and 'start_time' in values and v <= values['start_time']:
+            raise ValueError('recurrence_end_date must be after start_time')
+        return v
+
+class RecurringEventUpdate(EventUpdate):
+    recurrence_pattern: Optional[str] = None
+    recurrence_end_date: Optional[datetime] = None
+
+    @validator('recurrence_pattern')
+    def valid_recurrence_pattern(cls, v):
+        if v is not None:
+            valid_patterns = ['daily', 'weekly', 'monthly', 'yearly']
+            if v not in valid_patterns:
+                raise ValueError(f'recurrence_pattern must be one of: {", ".join(valid_patterns)}')
+        return v

app/services/analytics.py

@@ -0,0 +1,131 @@
+from datetime import datetime, timedelta
+from ..db.database import db
+from ..utils.cache import cache
+from typing import Dict, List, Any
+
+class AnalyticsService:
+    @staticmethod
+    async def get_sales_analytics(start_date: datetime, end_date: datetime) -> Dict[str, Any]:
+        cache_key = f"sales_analytics:{start_date.date()}:{end_date.date()}"
+        cached_data = await cache.get_cache(cache_key)
+        if cached_data:
+            return cached_data
+
+        pipeline = [
+            {
+                "$match": {
+                    "created_at": {
+                        "$gte": start_date,
+                        "$lte": end_date
+                    },
+                    "status": {"$in": ["completed", "delivered"]}
+                }
+            },
+            {
+                "$group": {
+                    "_id": {"$dateToString": {"format": "%Y-%m-%d", "date": "$created_at"}},
+                    "total_sales": {"$sum": "$total_amount"},
+                    "order_count": {"$sum": 1}
+                }
+            },
+            {"$sort": {"_id": 1}}
+        ]
+
+        sales_data = await db.db["orders"].aggregate(pipeline).to_list(None)
+        result = {
+            "daily_sales": sales_data,
+            "total_revenue": sum(day["total_sales"] for day in sales_data),
+            "total_orders": sum(day["order_count"] for day in sales_data),
+            "average_order_value": sum(day["total_sales"] for day in sales_data) / 
+                                 (sum(day["order_count"] for day in sales_data) or 1)
+        }
+
+        await cache.set_cache(cache_key, result, expire=3600)  # Cache for 1 hour
+        return result
+
+    @staticmethod
+    async def get_product_analytics() -> Dict[str, Any]:
+        cache_key = "product_analytics"
+        cached_data = await cache.get_cache(cache_key)
+        if cached_data:
+            return cached_data
+
+        pipeline = [
+            {
+                "$unwind": "$products"
+            },
+            {
+                "$group": {
+                    "_id": "$products.product_id",
+                    "total_quantity": {"$sum": "$products.quantity"},
+                    "total_revenue": {
+                        "$sum": {
+                            "$multiply": ["$products.price", "$products.quantity"]
+                        }
+                    }
+                }
+            },
+            {
+                "$sort": {"total_revenue": -1}
+            },
+            {
+                "$limit": 10
+            }
+        ]
+
+        top_products = await db.db["orders"].aggregate(pipeline).to_list(None)
+        
+        # Get product details
+        for product in top_products:
+            product_detail = await db.db["products"].find_one({"_id": product["_id"]})
+            if product_detail:
+                product["name"] = product_detail["name"]
+                product["category"] = product_detail["category"]
+
+        result = {
+            "top_products": top_products,
+            "total_products": await db.db["products"].count_documents({}),
+            "low_stock_products": await db.db["products"].count_documents({"inventory_count": {"$lt": 10}})
+        }
+
+        await cache.set_cache(cache_key, result, expire=3600)  # Cache for 1 hour
+        return result
+
+    @staticmethod
+    async def get_customer_analytics() -> Dict[str, Any]:
+        cache_key = "customer_analytics"
+        cached_data = await cache.get_cache(cache_key)
+        if cached_data:
+            return cached_data
+
+        pipeline = [
+            {
+                "$group": {
+                    "_id": "$customer_id",
+                    "total_orders": {"$sum": 1},
+                    "total_spent": {"$sum": "$total_amount"},
+                    "last_order": {"$max": "$created_at"}
+                }
+            },
+            {
+                "$sort": {"total_spent": -1}
+            }
+        ]
+
+        customer_data = await db.db["orders"].aggregate(pipeline).to_list(None)
+        
+        result = {
+            "total_customers": len(customer_data),
+            "top_customers": customer_data[:10],
+            "average_customer_value": sum(c["total_spent"] for c in customer_data) / (len(customer_data) or 1),
+            "customer_segments": {
+                "high_value": len([c for c in customer_data if c["total_spent"] > 1000]),
+                "medium_value": len([c for c in customer_data if 500 <= c["total_spent"] <= 1000]),
+                "low_value": len([c for c in customer_data if c["total_spent"] < 500])
+            }
+        }
+
+        await cache.set_cache(cache_key, result, expire=3600)  # Cache for 1 hour
+        return result
+
+analytics = AnalyticsService()

app/services/backup.py

@@ -0,0 +1,179 @@
+import os
+import shutil
+import json
+import tarfile
+from datetime import datetime
+from typing import Dict, Any, List
+from bson import ObjectId
+from ..db.database import db
+from ..utils.logger import logger
+
+class BackupService:
+    def __init__(self):
+        self.backup_dir = "backups"
+        self._ensure_backup_dir()
+
+    def _ensure_backup_dir(self):
+        """Ensure backup directory exists"""
+        if not os.path.exists(self.backup_dir):
+            os.makedirs(self.backup_dir)
+
+    async def create_backup(self, include_files: bool = True) -> Dict[str, Any]:
+        """Create a new system backup"""
+        try:
+            timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+            backup_id = str(ObjectId())
+            backup_name = f"backup_{timestamp}_{backup_id}"
+            backup_path = os.path.join(self.backup_dir, backup_name)
+            
+            # Create backup directory
+            os.makedirs(backup_path, exist_ok=True)
+            
+            # Backup database collections
+            db_backup = {}
+            for collection in await db.db.list_collection_names():
+                docs = await db.db[collection].find().to_list(None)
+                db_backup[collection] = [
+                    {**doc, "_id": str(doc["_id"])}
+                    for doc in docs
+                ]
+            
+            # Save database backup
+            with open(os.path.join(backup_path, "database.json"), "w") as f:
+                json.dump(db_backup, f, default=str)
+            
+            # Backup files if requested
+            if include_files:
+                uploads_dir = "uploads"
+                if os.path.exists(uploads_dir):
+                    shutil.copytree(
+                        uploads_dir,
+                        os.path.join(backup_path, "uploads"),
+                        dirs_exist_ok=True
+                    )
+            
+            # Create archive
+            archive_path = f"{backup_path}.tar.gz"
+            with tarfile.open(archive_path, "w:gz") as tar:
+                tar.add(backup_path, arcname=os.path.basename(backup_path))
+            
+            # Clean up temporary directory
+            shutil.rmtree(backup_path)
+            
+            # Record backup in database
+            backup_info = {
+                "_id": backup_id,
+                "filename": f"{backup_name}.tar.gz",
+                "path": archive_path,
+                "created_at": datetime.utcnow(),
+                "size": os.path.getsize(archive_path),
+                "includes_files": include_files
+            }
+            
+            await db.db["backup_history"].insert_one(backup_info)
+            
+            return {
+                "id": backup_id,
+                "path": archive_path,
+                "size": backup_info["size"],
+                "created_at": backup_info["created_at"]
+            }
+            
+        except Exception as e:
+            logger.error(f"Backup creation failed: {str(e)}")
+            raise
+
+    async def restore_backup(self, backup_path: str) -> Dict[str, Any]:
+        """Restore system from a backup"""
+        try:
+            if not os.path.exists(backup_path):
+                raise FileNotFoundError("Backup file not found")
+            
+            # Create temporary restoration directory
+            restore_dir = f"restore_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}"
+            os.makedirs(restore_dir, exist_ok=True)
+            
+            # Extract archive
+            with tarfile.open(backup_path, "r:gz") as tar:
+                tar.extractall(restore_dir)
+            
+            backup_contents = os.listdir(restore_dir)[0]
+            backup_root = os.path.join(restore_dir, backup_contents)
+            
+            # Restore database
+            with open(os.path.join(backup_root, "database.json"), "r") as f:
+                db_backup = json.load(f)
+            
+            # Clear existing collections
+            for collection in await db.db.list_collection_names():
+                await db.db[collection].delete_many({})
+            
+            # Restore collections
+            for collection, docs in db_backup.items():
+                if docs:
+                    # Convert string IDs back to ObjectId
+                    for doc in docs:
+                        doc["_id"] = ObjectId(doc["_id"])
+                    await db.db[collection].insert_many(docs)
+            
+            # Restore files if present
+            uploads_source = os.path.join(backup_root, "uploads")
+            if os.path.exists(uploads_source):
+                if os.path.exists("uploads"):
+                    shutil.rmtree("uploads")
+                shutil.copytree(uploads_source, "uploads")
+            
+            # Clean up
+            shutil.rmtree(restore_dir)
+            
+            return {
+                "success": True,
+                "collections_restored": len(db_backup),
+                "files_restored": os.path.exists(uploads_source)
+            }
+            
+        except Exception as e:
+            logger.error(f"Backup restoration failed: {str(e)}")
+            raise
+        finally:
+            if os.path.exists(restore_dir):
+                shutil.rmtree(restore_dir)
+
+    async def list_backups(self) -> List[Dict[str, Any]]:
+        """List all available backups"""
+        try:
+            backups = await db.db["backup_history"].find().sort("created_at", -1).to_list(None)
+            return [
+                {
+                    "id": str(backup["_id"]),
+                    "filename": backup["filename"],
+                    "created_at": backup["created_at"],
+                    "size": backup["size"],
+                    "includes_files": backup["includes_files"]
+                }
+                for backup in backups
+            ]
+        except Exception as e:
+            logger.error(f"Failed to list backups: {str(e)}")
+            raise
+
+    async def delete_backup(self, backup_id: str) -> bool:
+        """Delete a backup"""
+        try:
+            backup = await db.db["backup_history"].find_one({"_id": backup_id})
+            if not backup:
+                return False
+            
+            # Delete the physical backup file
+            if os.path.exists(backup["path"]):
+                os.remove(backup["path"])
+            
+            # Remove from database
+            await db.db["backup_history"].delete_one({"_id": backup_id})
+            return True
+            
+        except Exception as e:
+            logger.error(f"Failed to delete backup: {str(e)}")
+            raise
+
+backup = BackupService()

app/services/calendar.py

@@ -0,0 +1,215 @@
+from datetime import datetime, timedelta
+from typing import List, Dict, Any, Optional
+from bson import ObjectId
+from ..db.database import db
+from ..utils.cache import cache
+from ..services.notifications import notifications
+
+class CalendarService:
+    async def create_event(
+        self,
+        user_id: str,
+        title: str,
+        description: str,
+        start_time: datetime,
+        end_time: datetime,
+        attendees: List[str] = None,
+        is_all_day: bool = False,
+        reminder_minutes: int = 30
+    ) -> Dict[str, Any]:
+        """Create a new calendar event"""
+        event = {
+            "user_id": user_id,
+            "title": title,
+            "description": description,
+            "start_time": start_time,
+            "end_time": end_time,
+            "attendees": attendees or [],
+            "is_all_day": is_all_day,
+            "reminder_minutes": reminder_minutes,
+            "status": "scheduled",
+            "created_at": datetime.utcnow()
+        }
+        
+        result = await db.db["events"].insert_one(event)
+        event["_id"] = result.inserted_id
+        
+        # Clear cache for affected users
+        cache_keys = [f"user_events:{user_id}"]
+        for attendee in attendees or []:
+            cache_keys.append(f"user_events:{attendee}")
+        
+        for key in cache_keys:
+            await cache.delete_cache(key)
+        
+        # Notify attendees
+        if attendees:
+            for attendee in attendees:
+                await notifications.create_notification(
+                    user_id=attendee,
+                    title=f"New Event Invitation: {title}",
+                    message=f"You have been invited to an event: {title}",
+                    notification_type="event_invitation",
+                    data={"event_id": str(result.inserted_id)}
+                )
+        
+        return event
+
+    async def get_user_events(
+        self,
+        user_id: str,
+        start_date: datetime,
+        end_date: datetime,
+        include_attendee_events: bool = True
+    ) -> List[Dict[str, Any]]:
+        """Get events for a user within a date range"""
+        cache_key = f"user_events:{user_id}:{start_date.date()}:{end_date.date()}"
+        cached = await cache.get_cache(cache_key)
+        if cached:
+            return cached
+        
+        query = {
+            "$or": [
+                {"user_id": user_id},  # Events created by user
+                {"attendees": user_id} if include_attendee_events else {"_id": None}
+            ],
+            "start_time": {"$gte": start_date},
+            "end_time": {"$lte": end_date}
+        }
+        
+        cursor = db.db["events"].find(query).sort("start_time", 1)
+        events = await cursor.to_list(None)
+        
+        await cache.set_cache(cache_key, events, expire=300)  # Cache for 5 minutes
+        return events
+
+    async def update_event(
+        self,
+        event_id: str,
+        user_id: str,
+        update_data: Dict[str, Any]
+    ) -> Optional[Dict[str, Any]]:
+        """Update an event"""
+        if not ObjectId.is_valid(event_id):
+            return None
+        
+        event = await db.db["events"].find_one({
+            "_id": ObjectId(event_id),
+            "user_id": user_id  # Only creator can update
+        })
+        
+        if not event:
+            return None
+        
+        update_data["updated_at"] = datetime.utcnow()
+        
+        await db.db["events"].update_one(
+            {"_id": ObjectId(event_id)},
+            {"$set": update_data}
+        )
+        
+        # Clear cache for affected users
+        cache_keys = [f"user_events:{user_id}"]
+        for attendee in event.get("attendees", []):
+            cache_keys.append(f"user_events:{attendee}")
+        
+        for key in cache_keys:
+            await cache.delete_cache(key)
+        
+        # Notify attendees of changes
+        if "start_time" in update_data or "end_time" in update_data:
+            for attendee in event.get("attendees", []):
+                await notifications.create_notification(
+                    user_id=attendee,
+                    title=f"Event Updated: {event['title']}",
+                    message=f"An event you're attending has been updated",
+                    notification_type="event_update",
+                    data={"event_id": event_id}
+                )
+        
+        return await db.db["events"].find_one({"_id": ObjectId(event_id)})
+
+    async def delete_event(self, event_id: str, user_id: str) -> bool:
+        """Delete an event"""
+        if not ObjectId.is_valid(event_id):
+            return False
+        
+        event = await db.db["events"].find_one({
+            "_id": ObjectId(event_id),
+            "user_id": user_id  # Only creator can delete
+        })
+        
+        if not event:
+            return False
+        
+        result = await db.db["events"].delete_one({"_id": ObjectId(event_id)})
+        
+        if result.deleted_count > 0:
+            # Clear cache for affected users
+            cache_keys = [f"user_events:{user_id}"]
+            for attendee in event.get("attendees", []):
+                cache_keys.append(f"user_events:{attendee}")
+                # Notify attendees
+                await notifications.create_notification(
+                    user_id=attendee,
+                    title=f"Event Cancelled: {event['title']}",
+                    message=f"An event you were attending has been cancelled",
+                    notification_type="event_cancellation",
+                    data={"event_id": event_id}
+                )
+            
+            for key in cache_keys:
+                await cache.delete_cache(key)
+            
+            return True
+        return False
+
+    async def respond_to_event(
+        self,
+        event_id: str,
+        user_id: str,
+        response: str
+    ) -> bool:
+        """Respond to an event invitation"""
+        if not ObjectId.is_valid(event_id):
+            return False
+        
+        valid_responses = ["accepted", "declined", "maybe"]
+        if response not in valid_responses:
+            return False
+        
+        event = await db.db["events"].find_one({
+            "_id": ObjectId(event_id),
+            "attendees": user_id
+        })
+        
+        if not event:
+            return False
+        
+        # Update response in attendee list
+        await db.db["events"].update_one(
+            {"_id": ObjectId(event_id)},
+            {
+                "$set": {
+                    f"attendee_responses.{user_id}": response,
+                    "updated_at": datetime.utcnow()
+                }
+            }
+        )
+        
+        # Notify event creator
+        await notifications.create_notification(
+            user_id=event["user_id"],
+            title=f"Event Response: {event['title']}",
+            message=f"An attendee has {response} your event",
+            notification_type="event_response",
+            data={
+                "event_id": event_id,
+                "responder": user_id,
+                "response": response
+            }
+        )
+        
+        return True
+
+calendar = CalendarService()

app/services/maintenance.py

@@ -0,0 +1,217 @@
+import os
+import shutil
+import psutil
+from datetime import datetime, timedelta
+from typing import Dict, Any, Optional
+from sqlalchemy import select, delete, update, func
+from ..db.database import db
+from ..utils.logger import logger
+from ..core.config import settings
+from ..services.websocket import create_and_broadcast_notification
+from ..db.models import User, Order, Notification, Session
+
+class MaintenanceService:
+    async def cleanup_expired_sessions(self) -> int:
+        """Clean up expired sessions"""
+        try:
+            cutoff = datetime.utcnow() - timedelta(days=7)
+            async with db.session() as session:
+                stmt = delete(Session).where(Session.last_activity < cutoff)
+                result = await session.execute(stmt)
+                await session.commit()
+                return result.rowcount
+        except Exception as e:
+            logger.error(f"Error cleaning up sessions: {str(e)}")
+            return 0
+
+    async def archive_old_data(self) -> Dict[str, int]:
+        """Archive old data"""
+        try:
+            cutoff = datetime.utcnow() - timedelta(days=365)
+            archived = {}
+
+            async with db.session() as session:
+                # Archive old orders
+                order_stmt = update(Order).where(
+                    Order.created_at < cutoff,
+                    Order.status.in_(["completed", "cancelled"])
+                ).values(archived=True)
+                order_result = await session.execute(order_stmt)
+                archived["orders"] = order_result.rowcount
+
+                # Archive old notifications
+                notif_stmt = update(Notification).where(
+                    Notification.created_at < cutoff,
+                    Notification.read == True
+                ).values(archived=True)
+                notif_result = await session.execute(notif_stmt)
+                archived["notifications"] = notif_result.rowcount
+
+                await session.commit()
+                return archived
+
+        except Exception as e:
+            logger.error(f"Error archiving old data: {str(e)}")
+            return None
+
+    async def check_system_health(self) -> Dict[str, Any]:
+        """Check system health"""
+        try:
+            async with db.session() as session:
+                # Check database connection by running a simple query
+                await session.execute(select(func.now()))
+
+                # Get database size (using psutil for disk stats)
+                disk = psutil.disk_usage('/')
+                total_space = disk.total / (1024 * 1024 * 1024)  # GB
+                free_space = disk.free / (1024 * 1024 * 1024)  # GB
+                
+                health_data = {
+                    "status": "healthy",
+                    "database": {
+                        "connected": True
+                    },
+                    "disk": {
+                        "total_gb": total_space,
+                        "free_gb": free_space,
+                        "usage_percent": disk.percent
+                    },
+                    "timestamp": datetime.utcnow()
+                }
+
+                # Send alert if disk space is low
+                if free_space < 5:  # Less than 5GB free
+                    await create_and_broadcast_notification(
+                        user_id="admin",
+                        title="Low Disk Space Alert",
+                        message=f"Server is running low on disk space. Only {free_space:.2f}GB remaining.",
+                        notification_type="system_alert",
+                        data={"free_space_gb": free_space}
+                    )
+
+                return health_data
+
+        except Exception as e:
+            logger.error(f"Health check error: {str(e)}")
+            return {"status": "unhealthy", "error": str(e)}
+
+    async def monitor_system_resources(self) -> Dict[str, Any]:
+        """Monitor system resources"""
+        try:
+            async with db.session() as session:
+                # Get current active connections (using psutil for process stats)
+                process = psutil.Process()
+                open_files = process.open_files()
+                connections = len([f for f in open_files if 'socket' in str(f.path)])
+                
+                resources = {
+                    "database": {
+                        "connections": connections,
+                    },
+                    "system": {
+                        "cpu_percent": psutil.cpu_percent(),
+                        "memory_percent": psutil.virtual_memory().percent
+                    },
+                    "timestamp": datetime.utcnow()
+                }
+
+                # Alert if too many connections or high resource usage
+                if connections > settings.MAX_DB_CONNECTIONS * 0.9:
+                    await create_and_broadcast_notification(
+                        user_id="admin",
+                        title="High Database Connections",
+                        message=f"Database has {connections} active connections",
+                        notification_type="system_alert",
+                        data={"connections": connections}
+                    )
+
+                return resources
+
+        except Exception as e:
+            logger.error(f"Resource monitoring error: {str(e)}")
+            return {"error": str(e)}
+
+    async def perform_database_maintenance(self) -> Dict[str, Any]:
+        """Perform database maintenance tasks"""
+        try:
+            async with db.session() as session:
+                # Run ANALYZE on major tables
+                for table in [User, Order, Notification]:
+                    await session.execute(f"ANALYZE {table.__tablename__}")
+                
+                # Clean up any orphaned records
+                # For example, delete notifications for non-existent users
+                stmt = delete(Notification).where(
+                    ~Notification.user_id.in_(
+                        select(User.id)
+                    )
+                )
+                await session.execute(stmt)
+                await session.commit()
+                
+                return {"status": "success"}
+
+        except Exception as e:
+            logger.error(f"Database maintenance error: {str(e)}")
+            return {"error": str(e)}
+
+    async def rotate_log_files(self) -> None:
+        """Rotate log files"""
+        log_dir = "logs"
+        max_log_size = 10 * 1024 * 1024  # 10MB
+        
+        try:
+            for filename in os.listdir(log_dir):
+                filepath = os.path.join(log_dir, filename)
+                if os.path.getsize(filepath) > max_log_size:
+                    # Archive old log
+                    archive_name = f"{filename}.{datetime.now().strftime('%Y%m%d')}"
+                    shutil.move(filepath, os.path.join(log_dir, archive_name))
+                    
+                    # Create new log file
+                    open(filepath, 'a').close()
+                    logger.info(f"Rotated log file: {filename}")
+        except Exception as e:
+            logger.error(f"Log rotation error: {str(e)}")
+
+    async def manage_storage_quotas(self) -> Dict[str, Any]:
+        """Manage storage quotas and cleanup"""
+        try:
+            results = {
+                "warnings": [],
+                "cleaned": 0
+            }
+            
+            # Check and clean upload directories
+            upload_dirs = ["uploads/documents", "uploads/images"]
+            for directory in upload_dirs:
+                if os.path.exists(directory):
+                    total_size = sum(
+                        os.path.getsize(os.path.join(directory, f))
+                        for f in os.listdir(directory)
+                        if os.path.isfile(os.path.join(directory, f))
+                    ) / (1024 * 1024)  # Convert to MB
+                    
+                    if total_size > settings.MAX_UPLOAD_DIR_SIZE_MB:
+                        results["warnings"].append(
+                            f"Upload directory {directory} exceeds size limit"
+                        )
+            
+            # Clean up temporary files
+            temp_dirs = ["uploads/temp", "backups/temp"]
+            for directory in temp_dirs:
+                if os.path.exists(directory):
+                    # Remove files older than 24 hours
+                    cutoff = datetime.now() - timedelta(days=1)
+                    for filename in os.listdir(directory):
+                        filepath = os.path.join(directory, filename)
+                        if os.path.getctime(filepath) < cutoff.timestamp():
+                            os.remove(filepath)
+                            results["cleaned"] += 1
+            
+            return results
+        except Exception as e:
+            logger.error(f"Storage quota management error: {str(e)}")
+            return {"error": str(e)}
+
+maintenance = MaintenanceService()

app/services/notifications.py

@@ -0,0 +1,89 @@
+from typing import Dict, Any
+from datetime import datetime
+from ..db.database import db
+from ..core.config import settings
+from ..utils.cache import cache
+from ..services.websocket import create_and_broadcast_notification
+
+class NotificationService:
+    async def create_notification(
+        self,
+        user_id: str,
+        title: str,
+        message: str,
+        notification_type: str,
+        data: Dict[str, Any] = None
+    ):
+        """Create and store a notification"""
+        return await create_and_broadcast_notification(
+            user_id=user_id,
+            title=title,
+            message=message,
+            notification_type=notification_type,
+            data=data
+        )
+
+    async def get_user_notifications(
+        self,
+        user_id: str,
+        skip: int = 0,
+        limit: int = 50,
+        unread_only: bool = False
+    ):
+        """Get notifications for a user"""
+        cache_key = f"user_notifications:{user_id}"
+        if not unread_only:
+            cached = await cache.get_cache(cache_key)
+            if cached:
+                return cached
+
+        query = {"user_id": user_id}
+        if unread_only:
+            query["read"] = False
+
+        cursor = db.db["notifications"].find(query)\
+            .sort("created_at", -1)\
+            .skip(skip)\
+            .limit(limit)
+        
+        notifications = await cursor.to_list(length=limit)
+        
+        if not unread_only:
+            await cache.set_cache(cache_key, notifications, expire=300)  # Cache for 5 minutes
+        
+        return notifications
+
+    async def mark_as_read(self, notification_id: str, user_id: str):
+        """Mark a notification as read"""
+        result = await db.db["notifications"].update_one(
+            {"_id": notification_id, "user_id": user_id},
+            {"$set": {"read": True}}
+        )
+        
+        if result.modified_count > 0:
+            await cache.delete_cache(f"user_notifications:{user_id}")
+            return True
+        return False
+
+    async def mark_all_as_read(self, user_id: str):
+        """Mark all notifications as read for a user"""
+        result = await db.db["notifications"].update_many(
+            {"user_id": user_id, "read": False},
+            {"$set": {"read": True}}
+        )
+        
+        await cache.delete_cache(f"user_notifications:{user_id}")
+        return result.modified_count
+
+    async def delete_notification(self, notification_id: str, user_id: str):
+        """Delete a notification"""
+        result = await db.db["notifications"].delete_one(
+            {"_id": notification_id, "user_id": user_id}
+        )
+        
+        if result.deleted_count > 0:
+            await cache.delete_cache(f"user_notifications:{user_id}")
+            return True
+        return False
+
+notifications = NotificationService()

app/services/pos_client.py

@@ -0,0 +1,34 @@
+from typing import List
+from httpx import AsyncClient
+from pydantic import BaseModel
+from datetime import datetime
+from ..core.config import settings
+
+class OrderSchema(BaseModel):
+    id: str
+    branch_id: str
+    customer_id: str
+    items: List[dict]
+    total_amount: float
+    status: str
+    created_at: datetime
+    updated_at: datetime | None
+
+async def get_orders(branch_id: str) -> List[OrderSchema]:
+    """
+    Fetch orders from the POS system for a specific branch
+    
+    Args:
+        branch_id: The ID of the branch to fetch orders for
+        
+    Returns:
+        List[OrderSchema]: A list of orders from the POS system
+    """
+    async with AsyncClient() as client:
+        response = await client.get(
+            f"{settings.POS_API_URL}/internal/orders",
+            params={"branch_id": branch_id},
+            headers={"Authorization": f"Bearer {settings.SERVICE_TOKEN}"}
+        )
+        response.raise_for_status()
+        return [OrderSchema(**order) for order in response.json()]

app/services/scheduler.py

@@ -0,0 +1,243 @@
+from datetime import datetime, timedelta
+from typing import List, Dict, Any, Optional
+from bson import ObjectId
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.cron import CronTrigger
+from ..db.database import db
+from ..utils.cache import cache
+from ..utils.logger import logger
+from ..services.calendar import calendar
+from ..services.maintenance import maintenance
+
+class SchedulerService:
+    def __init__(self):
+        self.scheduler = AsyncIOScheduler()
+        self._setup_maintenance_jobs()
+
+    def _setup_maintenance_jobs(self):
+        """Setup all maintenance related scheduled jobs"""
+        # Daily database maintenance at 2 AM
+        self.scheduler.add_job(
+            maintenance.perform_database_maintenance,
+            CronTrigger(hour=2),
+            id="daily_db_maintenance",
+            replace_existing=True
+        )
+
+        # Session cleanup every 6 hours
+        self.scheduler.add_job(
+            maintenance.cleanup_expired_sessions,
+            CronTrigger(hour="*/6"),
+            id="session_cleanup",
+            replace_existing=True
+        )
+
+        # System health check every 15 minutes
+        self.scheduler.add_job(
+            maintenance.monitor_system_resources,
+            CronTrigger(minute="*/15"),
+            id="health_check",
+            replace_existing=True
+        )
+
+        # Daily backup at 1 AM
+        self.scheduler.add_job(
+            maintenance.perform_scheduled_backup,
+            CronTrigger(hour=1),
+            id="daily_backup",
+            replace_existing=True
+        )
+
+        # Daily log rotation at 3 AM
+        self.scheduler.add_job(
+            maintenance.rotate_log_files,
+            CronTrigger(hour=3),
+            id="log_rotation",
+            replace_existing=True
+        )
+
+        # Storage quota check every 2 hours
+        self.scheduler.add_job(
+            maintenance.manage_storage_quotas,
+            CronTrigger(hour="*/2"),
+            id="storage_quota_check",
+            replace_existing=True
+        )
+
+        # Monthly data archiving at 4 AM on the 1st of each month
+        self.scheduler.add_job(
+            maintenance.archive_old_data,
+            CronTrigger(day=1, hour=4),
+            id="monthly_archiving",
+            replace_existing=True
+        )
+
+    def start(self):
+        """Start the scheduler"""
+        try:
+            self.scheduler.start()
+            logger.info("Scheduler started successfully")
+        except Exception as e:
+            logger.error(f"Failed to start scheduler: {str(e)}")
+            raise
+
+    def shutdown(self):
+        """Shutdown the scheduler"""
+        try:
+            self.scheduler.shutdown()
+            logger.info("Scheduler shutdown successfully")
+        except Exception as e:
+            logger.error(f"Error during scheduler shutdown: {str(e)}")
+            raise
+
+    def get_jobs(self):
+        """Get all scheduled jobs"""
+        return [
+            {
+                "id": job.id,
+                "name": job.name,
+                "next_run_time": job.next_run_time.isoformat() if job.next_run_time else None,
+                "trigger": str(job.trigger)
+            }
+            for job in self.scheduler.get_jobs()
+        ]
+
+    async def create_recurring_event(
+        self,
+        user_id: str,
+        title: str,
+        description: str,
+        start_time: datetime,
+        end_time: datetime,
+        recurrence_pattern: str,  # daily, weekly, monthly, yearly
+        recurrence_end_date: Optional[datetime] = None,
+        attendees: List[str] = None,
+        reminder_minutes: int = 30
+    ) -> List[Dict[str, Any]]:
+        """Create recurring events based on pattern"""
+        events = []
+        current_start = start_time
+        current_end = end_time
+        duration = end_time - start_time
+
+        while True:
+            if recurrence_end_date and current_start > recurrence_end_date:
+                break
+
+            # Create individual event instance
+            event = await calendar.create_event(
+                user_id=user_id,
+                title=title,
+                description=description,
+                start_time=current_start,
+                end_time=current_end,
+                attendees=attendees,
+                reminder_minutes=reminder_minutes
+            )
+            events.append(event)
+
+            # Calculate next occurrence
+            if recurrence_pattern == "daily":
+                current_start += timedelta(days=1)
+            elif recurrence_pattern == "weekly":
+                current_start += timedelta(weeks=1)
+            elif recurrence_pattern == "monthly":
+                # Add one month (approximately)
+                if current_start.month == 12:
+                    current_start = current_start.replace(year=current_start.year + 1, month=1)
+                else:
+                    current_start = current_start.replace(month=current_start.month + 1)
+            elif recurrence_pattern == "yearly":
+                current_start = current_start.replace(year=current_start.year + 1)
+
+            current_end = current_start + duration
+
+        return events
+
+    async def update_recurring_event(
+        self,
+        event_id: str,
+        user_id: str,
+        update_data: Dict[str, Any],
+        update_future: bool = True
+    ) -> List[Dict[str, Any]]:
+        """Update a recurring event and optionally its future occurrences"""
+        # Get the original event
+        event = await db.db["events"].find_one({
+            "_id": ObjectId(event_id),
+            "user_id": user_id
+        })
+
+        if not event:
+            return []
+
+        # Update the current event
+        await calendar.update_event(event_id, user_id, update_data)
+
+        updated_events = [event]
+
+        # Update future occurrences if requested
+        if update_future:
+            future_events = await db.db["events"].find({
+                "recurrence_group": event.get("recurrence_group"),
+                "start_time": {"$gt": event["start_time"]},
+                "user_id": user_id
+            }).to_list(None)
+
+            for future_event in future_events:
+                await calendar.update_event(
+                    str(future_event["_id"]),
+                    user_id,
+                    update_data
+                )
+                updated_events.append(future_event)
+
+        return updated_events
+
+    async def delete_recurring_event(
+        self,
+        event_id: str,
+        user_id: str,
+        delete_future: bool = True
+    ) -> bool:
+        """Delete a recurring event and optionally its future occurrences"""
+        event = await db.db["events"].find_one({
+            "_id": ObjectId(event_id),
+            "user_id": user_id
+        })
+
+        if not event:
+            return False
+
+        # Delete the current event
+        await calendar.delete_event(event_id, user_id)
+
+        # Delete future occurrences if requested
+        if delete_future and event.get("recurrence_group"):
+            await db.db["events"].delete_many({
+                "recurrence_group": event["recurrence_group"],
+                "start_time": {"$gt": event["start_time"]},
+                "user_id": user_id
+            })
+
+        return True
+
+    async def get_upcoming_recurring_events(
+        self,
+        user_id: str,
+        days: int = 30
+    ) -> List[Dict[str, Any]]:
+        """Get upcoming recurring events for a user"""
+        start_date = datetime.utcnow()
+        end_date = start_date + timedelta(days=days)
+
+        events = await calendar.get_user_events(
+            user_id=user_id,
+            start_date=start_date,
+            end_date=end_date,
+            include_attendee_events=True
+        )
+
+        return sorted(events, key=lambda x: x["start_time"])
+
+scheduler = SchedulerService()

app/services/websocket.py

@@ -0,0 +1,80 @@
+from typing import List, Dict, Any
+from fastapi import WebSocket
+from datetime import datetime
+from ..db.database import db
+from ..utils.cache import cache
+from ..db.models import Notification
+
+# Store active WebSocket connections
+active_connections: List[WebSocket] = []
+
+async def connect(websocket: WebSocket):
+    """Accept a new WebSocket connection"""
+    await websocket.accept()
+    active_connections.append(websocket)
+
+async def disconnect(websocket: WebSocket):
+    """Remove a WebSocket connection"""
+    if websocket in active_connections:
+        active_connections.remove(websocket)
+
+async def broadcast_message(message: dict):
+    """Broadcast a message to all connected clients"""
+    disconnected = []
+    for connection in active_connections:
+        try:
+            await connection.send_json(message)
+        except:
+            disconnected.append(connection)
+    
+    # Clean up disconnected clients
+    for connection in disconnected:
+        await disconnect(connection)
+
+async def create_and_broadcast_notification(
+    user_id: str,
+    title: str,
+    message: str,
+    notification_type: str,
+    data: Dict[str, Any] = None
+) -> Dict[str, Any]:
+    """Create and broadcast a notification"""
+    async with db.session() as session:
+        # Create notification
+        notification = Notification(
+            user_id=user_id,
+            title=title,
+            message=message,
+            type=notification_type,
+            data=data,
+            created_at=datetime.utcnow(),
+            read=False
+        )
+        
+        # Store in database
+        session.add(notification)
+        await session.commit()
+        await session.refresh(notification)
+        
+        # Convert to dict for broadcasting
+        notification_dict = {
+            "id": str(notification.id),
+            "user_id": notification.user_id,
+            "title": notification.title,
+            "message": notification.message,
+            "type": notification.type,
+            "data": notification.data,
+            "created_at": notification.created_at.isoformat(),
+            "read": notification.read
+        }
+        
+        # Broadcast to connected clients
+        await broadcast_message({
+            "type": "notification",
+            "data": notification_dict
+        })
+        
+        # Clear user's notification cache
+        await cache.delete_cache(f"user_notifications:{user_id}")
+        
+        return notification_dict