rag_memecoin_update_workflow.py

"""
RAG Memecoin Update Workflow

Orchestrates the complete update pipeline:
1. MemoryMemecoinInputSource - Single-item input
2. RAGMemecoinUpdateProcessor - Validation + re-embedding
3. MemecoinUpdateExecutor - Atomic vector DB updates

Returns success status after execution.
"""

import asyncio
import logging
from typing import Dict, Optional

from src.domain.input_source.memory_memecoin_input_source import (
    MemoryMemecoinInputSource,
)
from src.domain.processor.rag_memecoin_update_processor import (
    RAGMemecoinUpdateProcessor,
)
from src.domain.executor.memecoin_update_executor import MemecoinUpdateExecutor
from src.domain.workflow.default_workflow import DefaultWorkflow
from src.services.ai.clip_embedding_service import CLIPEmbeddingService
from src.vector_store.memecoin_store import MemecoinVectorStore

logger = logging.getLogger(__name__)


class RAGMemecoinUpdateWorkflow:
    """
    Workflow for updating existing memecoins in vector database.

    Flow:
    1. MemoryMemecoinInputSource emits single MemecoinReadEvent
    2. RAGMemecoinUpdateProcessor validates and re-embeds captions
    3. MemecoinUpdateExecutor atomically updates all 5 vector DB collections
    4. Returns success/failure status
    """

    def __init__(
        self,
        clip_service: CLIPEmbeddingService,
        vector_store: MemecoinVectorStore,
    ):
        self.clip_service = clip_service
        self.vector_store = vector_store
        self.workflow: Optional[DefaultWorkflow] = None
        self.input_source: Optional[MemoryMemecoinInputSource] = None

    async def run(
        self, memecoin_data: Dict, token_address: str
    ) -> tuple[bool, str]:
        """
        Run the update workflow for a single memecoin.

        Args:
            memecoin_data: Dict with updated memecoin data (must include caption_structured)
            token_address: Token address to update

        Returns:
            tuple[bool, str]: (success, message)
        """
        try:
            logger.info(
                f"🚀 Starting RAGMemecoinUpdateWorkflow for {token_address}"
            )

            # Create input source
            self.input_source = MemoryMemecoinInputSource(memecoin_data)

            # Create processor
            processor = RAGMemecoinUpdateProcessor(
                clip_service=self.clip_service,
                token_address=token_address,
            )

            # Create executor
            executor = MemecoinUpdateExecutor(vector_store=self.vector_store)

            # Create workflow
            self.workflow = DefaultWorkflow(
                input_source=self.input_source,
                processor=processor,
                executor=executor,
            )

            # Initialize and run workflow
            await self.workflow.initialize()

            # Start workflow (runs until input source stops)
            await self.workflow.run_indefinitely()

            # Verify that processing actually succeeded by checking executor execution count
            # If validation failed or processing was terminated, executor.execute() was never called
            if executor.execution_count == 0:
                logger.error(
                    f"❌ RAGMemecoinUpdateWorkflow validation failed for {token_address}"
                )
                return False, f"Validation failed or processing incomplete for {token_address}"

            logger.info(
                f"✅ RAGMemecoinUpdateWorkflow completed for {token_address}"
            )
            return True, f"Successfully updated {token_address}"

        except Exception as e:
            error_msg = f"Update workflow failed for {token_address}: {e}"
            logger.error(f"❌ {error_msg}")
            return False, error_msg

    async def cleanup(self):
        """Cleanup workflow resources"""
        try:
            if self.workflow:
                await self.workflow.stop()
            logger.info("🧹 Update workflow cleaned up")
        except Exception as e:
            logger.error(f"❌ Error during cleanup: {e}")