Fixing collection deletion

trustgraph-flow/trustgraph/storage/doc_embeddings/milvus/write.py

@@ -60,6 +60,12 @@ class Processor(DocumentEmbeddingsStoreService):
             metrics=storage_response_metrics,
         )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def store_document_embeddings(self, message):
 
         for emb in message.chunks:

trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py

@@ -115,6 +115,12 @@ class Processor(DocumentEmbeddingsStoreService):
                 "Gave up waiting for index creation"
             )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def store_document_embeddings(self, message):
 
         for emb in message.chunks:

trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py

@@ -71,6 +71,14 @@ class Processor(DocumentEmbeddingsStoreService):
                 metrics=storage_response_metrics,
             )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        if hasattr(self, 'storage_request_consumer'):
+            await self.storage_request_consumer.start()
+        if hasattr(self, 'storage_response_producer'):
+            await self.storage_response_producer.start()
+
     async def store_document_embeddings(self, message):
 
         for emb in message.chunks:

trustgraph-flow/trustgraph/storage/graph_embeddings/milvus/write.py

@@ -60,6 +60,12 @@ class Processor(GraphEmbeddingsStoreService):
             metrics=storage_response_metrics,
         )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def store_graph_embeddings(self, message):
 
         for entity in message.entities:

trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py

@@ -115,6 +115,12 @@ class Processor(GraphEmbeddingsStoreService):
                 "Gave up waiting for index creation"
             )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def store_graph_embeddings(self, message):
 
         for entity in message.entities:

trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py

@@ -96,6 +96,14 @@ class Processor(GraphEmbeddingsStoreService):
 
         return cname
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        if hasattr(self, 'storage_request_consumer'):
+            await self.storage_request_consumer.start()
+        if hasattr(self, 'storage_response_producer'):
+            await self.storage_response_producer.start()
+
     async def store_graph_embeddings(self, message):
 
         for entity in message.entities:

trustgraph-flow/trustgraph/storage/objects/cassandra/write.py

@@ -340,6 +340,12 @@ class Processor(FlowProcessor):
             logger.warning(f"Failed to convert value {value} to type {field_type}: {e}")
             return str(value)
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def on_object(self, msg, consumer, flow):
         """Process incoming ExtractedObject and store in Cassandra"""
         
@@ -459,7 +465,7 @@ class Processor(FlowProcessor):
                     message=str(e)
                 )
             )
-            await self.send("storage-response", response)
+            await self.storage_response_producer.send(response)
 
     async def delete_collection(self, user: str, collection: str):
         """Delete all data for a specific collection"""

trustgraph-flow/trustgraph/storage/triples/cassandra/write.py

@@ -117,6 +117,12 @@ class Processor(TriplesStoreService):
                 t.o.value
             )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def on_storage_management(self, message):
         """Handle storage management requests"""
         logger.info(f"Storage management request: {message.operation} for {message.user}/{message.collection}")

trustgraph-flow/trustgraph/storage/triples/falkordb/write.py

@@ -185,6 +185,12 @@ class Processor(TriplesStoreService):
             help=f'FalkorDB database (default: {default_database})'
         )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def on_storage_management(self, message):
         """Handle storage management requests"""
         logger.info(f"Storage management request: {message.operation} for {message.user}/{message.collection}")

trustgraph-flow/trustgraph/storage/triples/memgraph/write.py

@@ -317,6 +317,12 @@ class Processor(TriplesStoreService):
             help=f'Memgraph database (default: {default_database})'
         )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def on_storage_management(self, message):
         """Handle storage management requests"""
         logger.info(f"Storage management request: {message.operation} for {message.user}/{message.collection}")

trustgraph-flow/trustgraph/storage/triples/neo4j/write.py

@@ -268,6 +268,12 @@ class Processor(TriplesStoreService):
             help=f'Neo4j database (default: {default_database})'
         )
 
+    async def start(self):
+        """Start the processor and its storage management consumer"""
+        await super().start()
+        await self.storage_request_consumer.start()
+        await self.storage_response_producer.start()
+
     async def on_storage_management(self, message):
         """Handle storage management requests"""
         logger.info(f"Storage management request: {message.operation} for {message.user}/{message.collection}")