Services Layer: Domain Business Logic

Purpose & Architecture

The services/ layer contains business logic for each domain entity:

• CommonService Base Class: All services inherit from CommonService, providing shared methods: pagination, tenant filtering, error handling, logging.
• 15+ Domain Services: Each service manages one or more related entities (DocumentService manages Document + file uploads, DialogService manages Dialog definitions, etc.).
• State Transition Management: Services handle complex state machines (e.g., Task: pending → processing → done/error). Handlers never directly update entity state.
• Atomic Operations: Services coordinate multi-step operations (create document + queue task, create conversation + send welcome message) atomically.
• Soft Delete Coordination: Services enforce soft-delete semantics globally (queries filter deleted_at.is_null(), restore operations reset deleted_at).

Common Service Base Class

All services inherit from CommonService in common_service.py:

class CommonService:
    '''Base class for all domain services.'''
    
    def __init__(self):
        self._logger = logging.getLogger(self.__class__.__name__)
    
    def get_by_id(self, id: str, tenant_id: str = None):
        '''Retrieve entity by ID with optional tenant check.'''
        entity = Entity.get_by_id(id)
        if tenant_id and entity.tenant_id != tenant_id:
            raise TenantMismatch(f"Entity {id} not in tenant {tenant_id}")
        return entity
    
    def list_by_tenant(self, tenant_id: str, offset=0, limit=50):
        '''Paginated list for tenant.'''
        return Entity.select().where(
            Entity.tenant_id == tenant_id,
            Entity.deleted_at.is_null()
        ).offset(offset).limit(limit)
    
    def delete(self, id: str, tenant_id: str):
        '''Soft delete with tenant check.'''
        entity = self.get_by_id(id, tenant_id)
        entity.deleted_at = datetime.now()
        entity.save()
        return entity

Domain Services Catalog

Service Design Patterns

Pattern 1: CRUD with Tenant Isolation

class DialogService(CommonService):
    def create(self, tenant_id: str, name: str, description: str = ""):
        if not name:
            raise ValidationError("name required")
        
        dialog = Dialog.create(
            tenant_id=tenant_id,
            name=name,
            description=description,
            created_by=g.user_id
        )
        self._logger.info(f"Created dialog {dialog.id} in tenant {tenant_id}")
        return dialog
    
    def get(self, dialog_id: str, tenant_id: str = None) -> Dialog:
        dialog = Dialog.get_by_id(dialog_id)
        if tenant_id and dialog.tenant_id != tenant_id:
            raise TenantMismatch(f"Dialog {dialog_id} not in tenant {tenant_id}")
        return dialog
    
    def list_by_tenant(self, tenant_id: str) -> List[Dialog]:
        return Dialog.select().where(
            Dialog.tenant_id == tenant_id,
            Dialog.deleted_at.is_null()
        ).order_by(Dialog.created_at.desc())

Pattern 2: State Transition (Task Lifecycle)

class TaskService(CommonService):
    def create(self, doc_id: str, task_type: str, tenant_id: str) -> Task:
        doc = DocumentService().get(doc_id, tenant_id)
        
        task = Task.create(
            document_id=doc_id,
            tenant_id=tenant_id,
            status="pending",
            task_type=task_type,
            progress_json={"percent": 0, "stage": "queued"}
        )
        
        # Queue task in Redis for background executor
        self._enqueue_task(task)
        return task
    
    def update_progress(self, task_id: str, progress: dict, tenant_id: str):
        task = self.get(task_id, tenant_id)
        task.progress_json = progress
        task.save()
        
        # Notify WebSocket subscribers of progress update
        self._broadcast_progress(task_id, progress)
    
    def mark_done(self, task_id: str, tenant_id: str, result: dict = None):
        task = self.get(task_id, tenant_id)
        task.status = "done"
        task.progress_json = {"percent": 100, "stage": "complete", "result": result}
        task.save()
        
        # Clean up temporary files, send completion notification
        self._finalize_task(task)

Pattern 3: Complex Mutation (Document Upload)

class DocumentService(CommonService):
    def upload(self, file: FileStorage, tenant_id: str, kb_id: str = None):
        '''Upload document: create Document row, save file, create Task, return document.'''
        
        # Validate inputs
        if not file or not file.filename:
            raise ValidationError("file required")
        
        # Create Document row
        doc = Document.create(
            tenant_id=tenant_id,
            name=file.filename,
            file_type=self._get_file_type(file.filename),
            created_by=g.user_id
        )
        
        # Save file to blob storage (S3 or local)
        file_path = self._save_file(file, doc.id)
        doc.file_path = file_path
        doc.save()
        
        # Add to knowledge base if provided
        if kb_id:
            KnowledgebaseService().add_document(kb_id, doc.id, tenant_id)
        
        # Create processing task
        task = TaskService().create(doc.id, "parse_and_embed", tenant_id)
        
        return {"document": doc, "task": task}
    
    def _save_file(self, file: FileStorage, doc_id: str) -> str:
        '''Save file to blob storage; return file path.'''
        # Implementation: write to S3 or local storage
        pass

Joint Services for Cross-Domain Coordination

Some operations require coordinating multiple services atomically. Joint services in joint_services/ manage these:

Error Handling & Logging

Services implement consistent error handling and logging:

import logging
from common.exceptions import (
    ValidationError, TenantMismatch, EntityNotFound, InternalError
)

class DocumentService(CommonService):
    def create(self, tenant_id: str, name: str):
        try:
            if not name or len(name) > 255:
                raise ValidationError("name must be 1-255 chars")
            
            doc = Document.create(
                tenant_id=tenant_id,
                name=name,
                created_by=g.user_id
            )
            self._logger.info(f"Created document {doc.id}")
            return doc
            
        except ValidationError:
            raise  # Re-raise domain exceptions
        except IntegrityError as e:
            self._logger.warning(f"DB integrity error: {e}")
            raise ValidationError("name already exists in tenant")
        except Exception as e:
            self._logger.exception(f"Unexpected error creating document: {e}")
            raise InternalError("Failed to create document")

Service Dependencies & Cross-References

Services can call other services for cross-domain operations:

• DocumentService → FileService: Upload document calls FileService.upload() to save file.
• DocumentService → TaskService: Upload document calls TaskService.create() to queue processing task.
• ConversationService → MessageService: Start conversation calls MessageService.create() to send welcome message.
• DialogService → LLMService: Get dialog config calls LLMService.get() to fetch associated LLM model.
• TaskService → ChunkService: Document parsing task calls ChunkService.batch_create() to insert chunks.

Tenant Isolation in Services

Every service method that accesses or modifies data must enforce tenant isolation:

• Check tenant_id on read: When retrieving entity, verify tenant_id matches request's tenant_id.
• Enforce tenant_id on create: Always set entity.tenant_id = request.tenant_id; never trust user input.
• Filter by tenant_id on list: All list queries include WHERE tenant_id = ? filter.
• Validate tenant_id on mutation: Before update/delete, check tenant_id matches.

Performance Considerations

• Connection Pool: Services use shared connection pool (10–20). Keep transactions short; avoid long-running DB work in services.
• Query Optimization: List operations paginate results (default limit 50). Add indexes on tenant_id, created_by, deleted_at.
• Caching: Services can implement caching for slowly-changing data (LLM models, KnowledgeBases). Use Redis with TTL.
• Deadlock Retry: Services that perform multi-table updates can retry on deadlock with exponential backoff.

Security Implications

• No SQL Injection: Services use Peewee ORM; all queries are parameterized.
• Tenant Data Leaks: Missing tenant_id check = critical vulnerability. Code review must verify every service method.
• Privilege Escalation: Services don't enforce role-based access control (admin vs user). That's the API layer's responsibility. Services assume caller is authorized.
• Credential Handling: Services that work with secrets (LLM API keys) must encrypt at rest, decrypt only when needed, never log.

Future Enhancements

• Event-Driven Architecture: Add event publishing (service creates Document → emits DocumentCreated event → other services listen). Loosens coupling.
• Service Versioning: Support multiple service implementations (DocumentService v1 vs v2) for gradual migration.
• Service Composition: Build higher-level "use case" services that orchestrate multiple domain services.
• Audit Logging: Log all mutations (who, what, when, why) for compliance and debugging.
• Transaction Sagas: For multi-step workflows (upload doc → parse → embed → index), implement saga pattern for failure recovery.