feat(queue): integrate redis queue and 2-minute paced ai consumer worker

.env.example

@@ -37,3 +37,5 @@ METRICS_PORT=8000
 MEDIA_DIR=/app/data/media
 GRAFANA_USER=admin
 GRAFANA_PASSWORD=admin
+REDIS_URL=redis://copykar_redis:6379/0
+AI_PROCESSING_INTERVAL_SECONDS=120

core/queue.py

@@ -0,0 +1,42 @@
+import os
+import logging
+from typing import Optional
+import redis.asyncio as redis
+
+logger = logging.getLogger(__name__)
+
+REDIS_URL = os.getenv("REDIS_URL", "redis://copykar_redis:6379/0" if os.path.exists("/app") else "redis://localhost:6379/0")
+
+class RedisQueue:
+    def __init__(self, redis_url: Optional[str] = None, queue_key: str = "copykar:queue:incoming"):
+        self.redis_url = redis_url or REDIS_URL
+        self.queue_key = queue_key
+        self.client: Optional[redis.Redis] = None
+
+    async def connect(self):
+        if not self.client:
+            self.client = redis.from_url(self.redis_url, decode_responses=True)
+            await self.client.ping()
+            logger.info(f"Connected to Redis at {self.redis_url}")
+
+    async def push(self, post_id: int):
+        if not self.client:
+            await self.connect()
+        await self.client.rpush(self.queue_key, str(post_id))
+        logger.info(f"Enqueued post ID {post_id} to Redis queue [{self.queue_key}]")
+
+    async def pop(self) -> Optional[int]:
+        if not self.client:
+            await self.connect()
+        val = await self.client.lpop(self.queue_key)
+        return int(val) if val else None
+
+    async def qsize(self) -> int:
+        if not self.client:
+            await self.connect()
+        return await self.client.llen(self.queue_key)
+
+    async def close(self):
+        if self.client:
+            await self.client.aclose()
+            logger.info("Redis connection closed.")

docker-compose.yml

@@ -17,6 +17,20 @@ services:
       timeout: 5s
       retries: 5
 
+  redis:
+    image: redis:7-alpine
+    container_name: copykar_redis
+    restart: unless-stopped
+    ports:
+      - "6379:6379"
+    volumes:
+      - redis_data:/data
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 5s
+      timeout: 3s
+      retries: 5
+
   copykar:
     image: copykar:latest
     build:
@@ -27,6 +41,8 @@ services:
     depends_on:
       postgres:
         condition: service_healthy
+      redis:
+        condition: service_healthy
     ports:
       - "8000:8000"
     env_file:
@@ -65,6 +81,7 @@ services:
 
 volumes:
   postgres_data:
+  redis_data:
   prometheus_data:
   grafana_data:
   copykar_data:

main.py

@@ -8,6 +8,8 @@ from db.database import init_db, close_db_pool
 from db.repository import Repository
 from core.metrics import start_metrics_server
 from core.llm import LLMClient
+from core.queue import RedisQueue
+from services.queue_consumer import QueueConsumerService
 from services.ai_processor import AIProcessor
 from services.collector import CollectorService
 from services.admin_bot import AdminBotService
@@ -30,10 +32,13 @@ async def main():
     metrics_port = int(os.getenv("METRICS_PORT", "8000"))
     start_metrics_server(metrics_port)
 
-    # 2. Initialize Database schema
+    # 2. Initialize Database schema & Redis Queue
     await init_db()
     logger.info("Database schema initialized.")
 
+    redis_queue = RedisQueue()
+    await redis_queue.connect()
+
     repo = Repository()
     llm = LLMClient()
 
@@ -50,13 +55,15 @@ async def main():
         return post
     ai_processor.process_post = process_and_notify
 
-    collector = CollectorService(repo=repo, ai_processor=ai_processor)
+    collector = CollectorService(repo=repo, ai_processor=ai_processor, queue=redis_queue)
     admin_bot.set_collector(collector)
     publisher = PublisherService(repo=repo)
+    queue_consumer = QueueConsumerService(queue=redis_queue, ai_processor=ai_processor)
 
     # 4. Start all services
     await admin_bot.start()
     await collector.start(notify_fn=admin_bot.notify_admins)
+    await queue_consumer.start()
     await publisher.start()
 
     logger.info("All Copykar services are active and running.")
@@ -78,8 +85,10 @@ async def main():
     finally:
         logger.info("Shutting down Copykar services...")
         await collector.stop()
+        await queue_consumer.stop()
         await publisher.stop()
         await admin_bot.stop()
+        await redis_queue.close()
         await close_db_pool()
         logger.info("Copykar cleanly shut down.")
 

requirements.txt

@@ -4,3 +4,4 @@ asyncpg==0.31.0
 prometheus-client==0.26.0
 httpx==0.28.1
 python-socks==3.0.0
+redis==8.1.0

services/admin_bot.py

@@ -165,9 +165,11 @@ class AdminBotService:
             approved = len(await self.repo.get_posts_by_status("approved", limit=1000))
             published = len(await self.repo.get_posts_by_status("published", limit=1000))
             rejected = len(await self.repo.get_posts_by_status("rejected", limit=1000))
+            redis_q = await self.collector.queue.qsize() if (self.collector and self.collector.queue) else 0
 
             text = (
                 "📊 <b>Copykar Fleet Metrics</b>\n\n"
+                f"• 📥 <b>Redis Incoming Queue:</b> {redis_q} (Pacing: 1 post / 2m)\n"
                 f"• ⏳ <b>Pending AI:</b> {pending_ai}\n"
                 f"• 📋 <b>Pending Review:</b> {pending_review}\n"
                 f"• 🚀 <b>Approved (In Queue):</b> {approved}\n"

services/collector.py

@@ -7,6 +7,7 @@ from telethon.tl.types import MessageMediaPhoto, MessageMediaDocument
 from db.repository import Repository
 from core.dedup import compute_content_hash, compute_file_hash
 from services.ai_processor import AIProcessor
+from core.queue import RedisQueue
 from core.metrics import COLLECTED_POSTS_TOTAL
 from core.proxy import get_telegram_proxy
 
@@ -20,6 +21,7 @@ class CollectorService:
         self,
         repo: Repository,
         ai_processor: AIProcessor,
+        queue: Optional[RedisQueue] = None,
         api_id: Optional[int] = None,
         api_hash: Optional[str] = None,
         phone: Optional[str] = None,
@@ -27,6 +29,7 @@ class CollectorService:
     ):
         self.repo = repo
         self.ai_processor = ai_processor
+        self.queue = queue
         self.api_id = api_id or int(os.getenv("API_ID", "0"))
         self.api_hash = api_hash or os.getenv("API_HASH", "")
         self.phone = phone or os.getenv("PHONE")
@@ -139,7 +142,10 @@ class CollectorService:
             if post_id:
                 COLLECTED_POSTS_TOTAL.labels(source_channel_id=str(chat_id)).inc()
                 logger.info(f"Collected new post ID {post_id} from channel {chat_id}")
-                await self.ai_processor.process_post(post_id)
+                if self.queue:
+                    await self.queue.push(post_id)
+                else:
+                    await self.ai_processor.process_post(post_id)
         except Exception as e:
             logger.error(f"Error handling message from {event.chat_id}: {e}", exc_info=True)
 
@@ -205,13 +211,16 @@ class CollectorService:
                     collected_count += 1
                     COLLECTED_POSTS_TOTAL.labels(source_channel_id=str(channel_id)).inc()
                     logger.info(f"Backfilled historical post ID {post_id} from {channel_id}")
-                    await self.ai_processor.process_post(post_id)
+                    if self.queue:
+                        await self.queue.push(post_id)
+                    else:
+                        await self.ai_processor.process_post(post_id)
                 else:
                     skipped_count += 1
 
             if progress_callback:
                 await progress_callback(
-                    f"✅ Scraped <b>{collected_count}</b> new historical posts from <code>{channel_id}</code> (Skipped {skipped_count} existing/empty)."
+                    f"✅ Scraped <b>{collected_count}</b> new posts from <code>{channel_id}</code> and queued in Redis! (Skipped {skipped_count} existing/empty)."
                 )
             return collected_count
         except Exception as e:

services/queue_consumer.py

@@ -0,0 +1,49 @@
+import os
+import asyncio
+import logging
+from typing import Optional
+from core.queue import RedisQueue
+from services.ai_processor import AIProcessor
+from core.metrics import QUEUE_POSTS_GAUGE
+
+logger = logging.getLogger(__name__)
+
+FETCH_INTERVAL_SECONDS = int(os.getenv("AI_PROCESSING_INTERVAL_SECONDS", "120"))
+
+class QueueConsumerService:
+    def __init__(self, queue: RedisQueue, ai_processor: AIProcessor, fetch_interval: int = FETCH_INTERVAL_SECONDS):
+        self.queue = queue
+        self.ai_processor = ai_processor
+        self.fetch_interval = fetch_interval
+        self._running = False
+        self._task: Optional[asyncio.Task] = None
+
+    async def start(self):
+        logger.info(f"Starting Redis Queue Consumer Service (interval: {self.fetch_interval}s / {self.fetch_interval // 60}m)...")
+        await self.queue.connect()
+        self._running = True
+        self._task = asyncio.create_task(self._consumer_loop())
+
+    async def _consumer_loop(self):
+        while self._running:
+            try:
+                qsize = await self.queue.qsize()
+                QUEUE_POSTS_GAUGE.labels(status="redis_incoming").set(qsize)
+
+                if qsize > 0:
+                    post_id = await self.queue.pop()
+                    if post_id:
+                        logger.info(f"Paced Consumer: processing post ID {post_id} from Redis queue (remaining: {qsize - 1})")
+                        await self.ai_processor.process_post(post_id)
+                        new_qsize = await self.queue.qsize()
+                        QUEUE_POSTS_GAUGE.labels(status="redis_incoming").set(new_qsize)
+            except Exception as e:
+                logger.error(f"Error in queue consumer loop: {e}", exc_info=True)
+
+            await asyncio.sleep(self.fetch_interval)
+
+    async def stop(self):
+        self._running = False
+        if self._task:
+            self._task.cancel()
+        logger.info("Redis Queue Consumer Service stopped.")