""" CQRS Sync Service. Keeps the read model (TaskInboxView) in sync with the write model (TableRecord). Handles archiving of completed records. """ import logging from datetime import timedelta from typing import Dict, List, Optional from uuid import UUID from django.db import transaction from django.utils import timezone from ..models import TableRecord, TableDefinition, RecordStatus from .models import TaskInboxView, TaskArchive logger = logging.getLogger(__name__) class ReadModelSyncService: """ Synchronizes the denormalized read model with the write model. Called via signal handlers or Celery tasks to maintain eventual consistency between write (TableRecord) and read (TaskInboxView) models. """ @staticmethod @transaction.atomic def sync_record(record: TableRecord) -> TaskInboxView: """ Sync a single record to the read model. Creates or updates the corresponding TaskInboxView entry. Args: record: TableRecord instance Returns: Updated TaskInboxView entry """ table = record.table # Build display title from first text-like column title = ReadModelSyncService._extract_title(record, table) description = ReadModelSyncService._extract_description(record, table) # Build display data (subset of fields for list view) display_data = ReadModelSyncService._build_display_data(record, table) # Get owner name owner_name = '' if record.owner: owner_name = ( getattr(record.owner, 'get_full_name', lambda: '')() or getattr(record.owner, 'email', '') ) inbox_entry, created = TaskInboxView.objects.update_or_create( record_id=record.id, defaults={ 'tenant': record.tenant, 'title': title, 'description': description, 'table_id': table.id, 'table_name': table.name, 'table_slug': table.slug, 'owner_id': record.owner_id, 'owner_name': owner_name, 'status': record.status, 'priority': record.data.get('priority', 50), 'due_date': record.data.get('due_date'), 'tags': record.data.get('tags', []), 'display_data': display_data, 'created_at': record.created_at, 'updated_at': record.updated_at, # Workflow fields 'workflow_instance_id': ( record.workflow_state.get('instance_id') if record.workflow_state else None ), 'workflow_name': ( record.workflow_state.get('workflow_name') if record.workflow_state else None ), }, ) action = 'Created' if created else 'Updated' logger.debug(f"{action} inbox view for record {record.id}") return inbox_entry @staticmethod def remove_record(record_id: UUID) -> bool: """Remove a record from the read model.""" deleted, _ = TaskInboxView.objects.filter(record_id=record_id).delete() return deleted > 0 @staticmethod def bulk_sync_table(table_id: UUID) -> int: """ Rebuild all read model entries for a table. Useful for initial population or when the schema changes. Args: table_id: TableDefinition UUID Returns: Number of records synced """ records = TableRecord.objects.filter( table_id=table_id, is_deleted=False, ).select_related('table', 'owner') count = 0 for record in records.iterator(): ReadModelSyncService.sync_record(record) count += 1 logger.info(f"Bulk synced {count} records for table {table_id}") return count @staticmethod def _extract_title(record: TableRecord, table: TableDefinition) -> str: """Extract a display title from the record data.""" # Try to find a 'title' or 'name' field for key in ['title', 'name', 'subject', 'label']: if key in record.data and record.data[key]: return str(record.data[key])[:500] # Fallback: use the first text column first_col = table.columns.filter( field_type__in=['text', 'textarea'] ).order_by('order').first() if first_col and first_col.slug in record.data: return str(record.data[first_col.slug])[:500] return f"{table.name} - {str(record.id)[:8]}" @staticmethod def _extract_description(record: TableRecord, table: TableDefinition) -> str: """Extract a description from the record data.""" for key in ['description', 'body', 'content', 'notes']: if key in record.data and record.data[key]: return str(record.data[key])[:2000] return '' @staticmethod def _build_display_data( record: TableRecord, table: TableDefinition, ) -> Dict: """Build a subset of record data for list view display.""" visible_columns = table.columns.filter( is_visible=True ).order_by('order')[:10] display = {} for col in visible_columns: if col.slug in record.data: display[col.slug] = { 'label': col.name, 'value': record.data[col.slug], 'type': col.field_type, } return display class ArchiveService: """ Manages archiving of completed/old records. Moves completed records from the active tables to the archive to keep the active tables lean and performant. """ @staticmethod @transaction.atomic def archive_record( record: TableRecord, reason: str = 'manual', user=None, ) -> TaskArchive: """ Archive a single record. Moves the record to TaskArchive and removes it from the active read model. Args: record: TableRecord to archive reason: Archive reason user: User performing the archive Returns: Created TaskArchive entry """ archive = TaskArchive.objects.create( tenant=record.tenant, original_record_id=record.id, table_id=record.table_id, table_name=record.table.name, owner_id=record.owner_id, data=record.data, status=record.status, workflow_state=record.workflow_state, version=record.version, archived_reason=reason, archived_by=user, original_created_at=record.created_at, original_updated_at=record.updated_at, ) # Remove from read model TaskInboxView.objects.filter(record_id=record.id).delete() # Soft delete the original record record.soft_delete() logger.info(f"Archived record {record.id} (reason: {reason})") return archive @staticmethod def auto_archive_completed( days_after_completion: int = 30, tenant=None, ) -> int: """ Auto-archive records completed more than N days ago. Args: days_after_completion: Days after completion to archive tenant: Optional tenant filter Returns: Number of records archived """ cutoff = timezone.now() - timedelta(days=days_after_completion) qs = TableRecord.objects.filter( status__in=[RecordStatus.APPROVED, RecordStatus.ARCHIVED], updated_at__lt=cutoff, is_deleted=False, ).select_related('table') if tenant: qs = qs.filter(tenant=tenant) count = 0 for record in qs.iterator(): try: ArchiveService.archive_record( record=record, reason='auto_completed', ) count += 1 except Exception as e: logger.error(f"Failed to auto-archive record {record.id}: {e}") logger.info(f"Auto-archived {count} records (cutoff: {cutoff})") return count @staticmethod @transaction.atomic def restore_from_archive( archive_id: UUID, ) -> TableRecord: """ Restore a record from the archive. Args: archive_id: TaskArchive UUID Returns: Restored TableRecord """ archive = TaskArchive.objects.get(pk=archive_id) # Restore or create the record try: record = TableRecord.objects.get(pk=archive.original_record_id) record.is_deleted = False record.deleted_at = None record.data = archive.data record.status = archive.status record.workflow_state = archive.workflow_state record.save() except TableRecord.DoesNotExist: record = TableRecord.objects.create( id=archive.original_record_id, tenant=archive.tenant, table_id=archive.table_id, owner_id=archive.owner_id, data=archive.data, status=archive.status, workflow_state=archive.workflow_state, version=archive.version, ) # Sync to read model ReadModelSyncService.sync_record(record) # Remove from archive archive.delete() logger.info(f"Restored record {record.id} from archive {archive_id}") return record