Upload 4 files

Dockerfile

@@ -1,4 +1,4 @@
-FROM mcr.microsoft.com/playwright/python:v1.58.0-jammy
+FROM mcr.microsoft.com/playwright/python:v1.40.0-jammy
 
 WORKDIR /app
 

hf_runner.py

@@ -1,6 +1,4 @@
 import os
-import sys
-import time
 import subprocess
 import threading
 from flask import Flask
@@ -12,23 +10,11 @@ def home():
     return "Worker is Running! 🚀"
 
 def run_worker():
-    """
-    Worker ko background mein run karta hai.
-    FIX #1: Auto-restart loop — agar worker crash ho toh 5s baad restart hoga
-    FIX #2: stdout/stderr properly piped taaki HF logs mein dikhe
-    """
-    while True:
-        print("[hf_runner] Starting worker.py...", flush=True)
-        result = subprocess.run(
-            ["python", "-u", "worker.py"],   # -u = unbuffered output
-            stdout=sys.stdout,
-            stderr=sys.stderr
-        )
-        print(f"[hf_runner] Worker exited with code {result.returncode}. Restarting in 5s...", flush=True)
-        time.sleep(5)
+    print("Starting Worker...")
+    subprocess.run(["python", "worker.py"])
 
 if __name__ == "__main__":
-    # Start worker in background thread with auto-restart
+    # Start worker in background
     threading.Thread(target=run_worker, daemon=True).start()
-    # Start dummy Flask server for Hugging Face health check (port 7860 required)
+    # Start dummy server for Hugging Face health check
     app.run(host="0.0.0.0", port=7860)

worker.py

@@ -29,25 +29,33 @@ import redis.asyncio as redis
 # Playwright imports
 from playwright.async_api import async_playwright, Page, Browser, BrowserContext
 
+# -----------------------------------------------------------------------------
+# Configuration
 # -----------------------------------------------------------------------------
 REDIS_HOST = os.getenv("REDIS_HOST", "13.235.87.209")
 REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
 REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
 
-LEAD_QUEUE = "lead_queue"
-PROCESSING_QUEUE = "processing_queue"
-RESULT_QUEUE = "result_queue"
-ACCOUNT_POOL_READY = "account_pool:ready"
+LEAD_QUEUE        = "lead_queue"
+PROCESSING_QUEUE  = "processing_queue"
+RESULT_QUEUE      = "result_queue"
+ACCOUNT_POOL_READY    = "account_pool:ready"
 ACCOUNT_POOL_COOLDOWN = "account_pool:cooldown"
-ACCOUNT_POOL_DONE = "account_pool:done_for_today"
-RETRY_COUNT_KEY = "retry_count"
-
-WORKER_TASKS = int(os.getenv("WORKER_THREADS", "5"))
-BATCH_SIZE = 10
-MAX_DAILY_LEADS = 40
-COOLDOWN_SECONDS = 3600  # 1 hour
-MAX_RETRIES = 3
-POLL_TIMEOUT = 5
+ACCOUNT_POOL_DONE     = "account_pool:done_for_today"
+RETRY_COUNT_KEY   = "retry_count"
+
+# Key that stores a SET of lead emails that are currently in-flight or already done.
+# This is the deduplication guard: a lead email is added here before processing
+# and removed only if it fails (so it can be retried). If it succeeds it stays
+# forever (within the day) so it cannot be sent again.
+SENT_LEADS_SET    = "leads:sent_today"
+
+WORKER_TASKS    = int(os.getenv("WORKER_THREADS", "5"))
+BATCH_SIZE      = 10          # leads per account session
+MAX_DAILY_LEADS = 40          # max leads per account per day  (4 batches × 10)
+COOLDOWN_SECONDS = 3600       # 1 hour between batches
+MAX_RETRIES     = 3
+POLL_TIMEOUT    = 5           # seconds to block on BRPOPLPUSH
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
 logger = logging.getLogger("worker")
@@ -57,48 +65,36 @@ _shutdown_requested = False
 # Machine identity (1 machine = 1 worker in dashboard)
 MACHINE_ID = socket.gethostname()
 
-# Shared state across all tasks on this machine
+# Shared stats across all coroutines on this machine
 _machine_stats = {
     "total_done": 0,
     "total_failed": 0,
-    "threads": {},  # task_id -> current_action
+    "threads": {},   # task_id -> current_action
 }
-
-# FIX #7: asyncio.Lock() module level par nahi, lazy init bhi nahi — main() ke andar bano
-_stats_lock: asyncio.Lock = None  # type: ignore
+_stats_lock = asyncio.Lock()
 
 def _signal_handler(signum, frame):
     global _shutdown_requested
     logger.info("Shutdown signal received. Draining worker pool...")
     _shutdown_requested = True
 
-signal.signal(signal.SIGINT, _signal_handler)
+signal.signal(signal.SIGINT,  _signal_handler)
 signal.signal(signal.SIGTERM, _signal_handler)
 
-# FIX #7: Redis pool eager init — module level par, race condition nahi hogi
-_redis_pool = redis.ConnectionPool(
-    host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD,
-    decode_responses=True, max_connections=50
-)
+_redis_pool = None
 
 def get_redis_pool():
+    global _redis_pool
+    if _redis_pool is None:
+        _redis_pool = redis.ConnectionPool(
+            host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD,
+            decode_responses=True, max_connections=50
+        )
     return _redis_pool
 
-
-# FIX #12: Redis startup connection check
-async def check_redis_connection():
-    """Startup mein Redis reachability check karo"""
-    client = redis.Redis(connection_pool=get_redis_pool())
-    try:
-        await client.ping()
-        logger.info(f"✅ Redis connected: {REDIS_HOST}:{REDIS_PORT}")
-    except Exception as e:
-        logger.critical(f"❌ Cannot connect to Redis at {REDIS_HOST}:{REDIS_PORT} — {e}")
-        sys.exit(1)
-    finally:
-        await client.aclose()
-
-
+# -----------------------------------------------------------------------------
+# Automation Engine
+# -----------------------------------------------------------------------------
 class AutomationEngine:
     """Robust Playwright automation engine"""
 
@@ -126,7 +122,7 @@ class AutomationEngine:
         self.browser = await self.playwright.chromium.launch(**browser_options)
         self.context = await self.browser.new_context()
         self.page = await self.context.new_page()
-        # Abort resource-heavy requests (images/fonts/media) for speed
+        # Abort resource-heavy requests
         await self.page.route("**/*", lambda route: (
             route.abort() if route.request.resource_type in ["image", "font", "media"]
             else route.continue_()
@@ -169,6 +165,7 @@ class AutomationEngine:
                 logger.error(f"Incorrect Temporary Access Pass for {email}")
                 return False
 
+            # Wait for "Keep your account secure" or "Authenticator" page to load
             await asyncio.sleep(4)
 
             # Handling potential security prompts
@@ -185,7 +182,7 @@ class AutomationEngine:
                 await self.page.wait_for_selector(selector, timeout=15000)
                 await self.page.locator(selector).click()
 
-                # Click the Email option
+                # Now click the Email option
                 await self.page.get_by_role("button", name="Email Receive a code to reset").click()
                 await asyncio.sleep(2)
             except Exception as e:
@@ -233,44 +230,111 @@ class AutomationEngine:
             return False, str(e)
 
 
+# -----------------------------------------------------------------------------
+# Worker Async Task Logic
+# -----------------------------------------------------------------------------
 class WorkerTask:
     def __init__(self, task_id: str):
-        self.task_id = task_id
-        self.client = redis.Redis(connection_pool=get_redis_pool())
-        self.leads_done = 0
+        self.task_id   = task_id
+        self.client    = redis.Redis(connection_pool=get_redis_pool())
+        self.leads_done   = 0
         self.leads_failed = 0
         self.current_action = "Starting"
 
-    async def _fetch_batch(self) -> List[Dict[str, Any]]:
-        """Fetch a batch of up to BATCH_SIZE leads from Redis."""
-        leads = []
-        try:
-            # FIX #8: brpoplpush deprecated — use blmove instead (Redis 6.2+)
-            raw_lead = await self.client.blmove(
-                LEAD_QUEUE, PROCESSING_QUEUE, "RIGHT", "LEFT", timeout=POLL_TIMEOUT
-            )
-            if not raw_lead:
-                return leads
-
-            leads.append({"_raw": raw_lead, "data": json.loads(raw_lead)})
+    # ------------------------------------------------------------------
+    # Fetch exactly BATCH_SIZE UNIQUE leads from the queue.
+    # Uses a Redis SET (SENT_LEADS_SET) to skip leads already processed
+    # today.  Each lead that passes the dedup check is atomically claimed
+    # via SETNX so no two threads can process the same email at once.
+    # ------------------------------------------------------------------
+    async def _fetch_unique_batch(self, account_email: str) -> List[Dict[str, Any]]:
+        """
+        Pop up to BATCH_SIZE × 3 items from the queue, de-duplicate against
+        SENT_LEADS_SET and claim them.  Returns at most BATCH_SIZE leads.
+
+        Claimed leads are added to SENT_LEADS_SET immediately so concurrent
+        threads cannot pick them up. Skipped duplicates are discarded (not
+        re-queued — they were already processed or in-flight).
+        """
+        unique_leads = []
+        seen_emails  = set()        # within this fetch call
+        skipped      = 0
 
-            # Non-blocking fetch for the rest of the batch
-            for _ in range(BATCH_SIZE - 1):
-                raw = await self.client.lmove(LEAD_QUEUE, PROCESSING_QUEUE, "RIGHT", "LEFT")
-                if not raw:
+        try:
+            # Block for the FIRST item
+            raw = await self.client.brpoplpush(LEAD_QUEUE, PROCESSING_QUEUE, timeout=POLL_TIMEOUT)
+            if not raw:
+                return []
+
+            candidates = [raw]
+
+            # Non-blocking fetch of extra candidates so we can fill a full batch
+            # even after dedup removes some items.
+            for _ in range(BATCH_SIZE * 3 - 1):
+                extra = await self.client.rpoplpush(LEAD_QUEUE, PROCESSING_QUEUE)
+                if not extra:
                     break
-                leads.append({"_raw": raw, "data": json.loads(raw)})
+                candidates.append(extra)
+
+            for raw_lead in candidates:
+                if len(unique_leads) >= BATCH_SIZE:
+                    # Put unused extras back to the front of lead_queue
+                    await self.client.rpush(LEAD_QUEUE, raw_lead)
+                    await self.client.lrem(PROCESSING_QUEUE, -1, raw_lead)
+                    continue
+
+                try:
+                    data = json.loads(raw_lead)
+                except Exception:
+                    # Malformed JSON — discard
+                    await self.client.lrem(PROCESSING_QUEUE, -1, raw_lead)
+                    continue
+
+                lead_email = data.get("email", "").strip().lower()
+                if not lead_email:
+                    await self.client.lrem(PROCESSING_QUEUE, -1, raw_lead)
+                    continue
+
+                # --- Deduplication ---
+                # 1. Already processed in this fetch call?
+                if lead_email in seen_emails:
+                    skipped += 1
+                    await self.client.lrem(PROCESSING_QUEUE, -1, raw_lead)
+                    continue
+
+                used_key = f"used:{account_email}"
+
+                # 2. Already sent by this account?
+                already_sent = await self.client.sismember(used_key, lead_email)
+                if already_sent:
+                    skipped += 1
+                    logger.debug(f"Skipping lead {lead_email} - already used by {account_email}")
+                    await self.client.lrem(PROCESSING_QUEUE, -1, raw_lead)
+                    continue
+
+                # Claim this lead for this account
+                claimed = await self.client.sadd(used_key, lead_email)
+                if not claimed:
+                    skipped += 1
+                    await self.client.lrem(PROCESSING_QUEUE, -1, raw_lead)
+                    continue
+
+                # Normalize email stored in data for consistency
+                data["email"] = lead_email
+                seen_emails.add(lead_email)
+                unique_leads.append({"_raw": raw_lead, "data": data})
 
         except Exception as e:
-            logger.error(f"Error fetching batch: {e}")
+            logger.error(f"Error fetching unique batch: {e}")
+
+        if skipped:
+            logger.info(f"[Dedup] Skipped {skipped} duplicate leads in this batch fetch.")
 
-        return leads
+        return unique_leads
 
-    async def _cleanup_and_push_results(self, leads: List[Dict[str, Any]], results: List[Dict[str, Any]]):
+    async def _cleanup_and_push_results(self, leads: List[Dict[str, Any]], results: List[Dict[str, Any]], account_email: str):
         if not leads:
             return
-        if len(leads) != len(results):
-            logger.error(f"Mismatch: {len(leads)} leads but {len(results)} results — some leads may be stuck!")
         try:
             pipe = self.client.pipeline()
             for lead, result in zip(leads, results):
@@ -280,6 +344,13 @@ class WorkerTask:
                 lead_id = lead["data"].get("id", str(uuid.uuid4()))
                 if result["success"]:
                     pipe.hdel(RETRY_COUNT_KEY, lead_id)
+                else:
+                    # On failure, un-claim from SENT_LEADS_SET so it can be
+                    # retried next time (unless max retries exceeded).
+                    retries = int(result.get("_retries", 0))
+                    if retries < MAX_RETRIES:
+                        lead_email = lead["data"].get("email", "")
+                        pipe.srem(f"used:{account_email}", lead_email)
 
             await pipe.execute()
         except Exception as e:
@@ -292,10 +363,12 @@ class WorkerTask:
             # Update shared machine stats
             async with _stats_lock:
                 _machine_stats["threads"][self.task_id] = self.current_action
-                _machine_stats["total_done"] = sum(t.leads_done for t in _all_worker_tasks)
+                _machine_stats["total_done"]   = sum(t.leads_done   for t in _all_worker_tasks)
                 _machine_stats["total_failed"] = sum(t.leads_failed for t in _all_worker_tasks)
 
-            # 1. Lease Account from Ready Pool
+            # ----------------------------------------------------------
+            # 1. Lease ONE account from the Ready pool (atomic SPOP)
+            # ----------------------------------------------------------
             email = await self.client.spop(ACCOUNT_POOL_READY)
             if not email:
                 self.current_action = "Idle"
@@ -303,65 +376,75 @@ class WorkerTask:
                 await asyncio.sleep(5)
                 continue
 
-            acc_key = f"account:{email}"
+            acc_key  = f"account:{email}"
             acc_pass = await self.client.hget(acc_key, "temp_pass")
 
-            # 2. Fetch Leads
+            if not acc_pass:
+                logger.warning(f"Account {email} has no temp_pass. Moving to pending_refresh.")
+                await self.client.sadd("account_pool:pending_refresh", email)
+                continue
+
+            # ----------------------------------------------------------
+            # 2. Fetch up to 10 UNIQUE leads
+            # ----------------------------------------------------------
             self.current_action = "Fetching Leads"
-            batch = await self._fetch_batch()
+            batch = await self._fetch_unique_batch(email)
+
             if not batch:
-                # No leads, return account to ready pool
+                # No leads — return account to ready pool and wait
                 await self.client.sadd(ACCOUNT_POOL_READY, email)
                 self.current_action = "Idle"
                 await asyncio.sleep(1)
                 continue
 
-            results = []
-            engine = AutomationEngine()
+            logger.info(f"Task {self.task_id}: Account={email}, Leads in batch={len(batch)}")
 
-            # FIX #2: login_success default False PEHLE try block se — UnboundLocalError prevent
-            login_success = False
+            results = []
+            engine  = AutomationEngine()
 
             try:
                 await engine.start(headless=True)
 
-                # 3. Login
+                # ----------------------------------------------------------
+                # 3. Login once per account session
+                # ----------------------------------------------------------
                 self.current_action = f"Logging in: {email}"
                 login_success = await engine.login_microsoft(email, acc_pass)
 
                 for item in batch:
-                    lead_data = item["data"]
-                    lead_id = lead_data.get("id", str(uuid.uuid4()))
-                    raw_data = item["_raw"]
+                    lead_data  = item["data"]
+                    lead_id    = lead_data.get("id", str(uuid.uuid4()))
+                    raw_data   = item["_raw"]
                     lead_email = lead_data.get("email")
 
                     self.current_action = f"Processing: {lead_email}"
 
                     if not login_success:
-                        # Fail entire batch if login fails
+                        # Login failed — requeue lead and unclaim from sent set
+                        pipe = self.client.pipeline()
+                        pipe.rpush(LEAD_QUEUE, raw_data)
+                        pipe.lrem(PROCESSING_QUEUE, -1, raw_data)
+                        pipe.srem(f"used:{email}", lead_email)   # ← un-claim so it can be retried
+                        await pipe.execute()
+
                         results.append({
                             "lead_id": lead_id,
                             "success": False,
                             "error": "login_failed",
                             "worker_id": self.task_id,
-                            "row_number": lead_data.get("row_number")
+                            "row_number": lead_data.get("row_number"),
+                            "_skip": True,
                         })
-
-                        # Requeue the raw_data
-                        pipe = self.client.pipeline()
-                        pipe.rpush(LEAD_QUEUE, raw_data)
-                        pipe.lrem(PROCESSING_QUEUE, -1, raw_data)
-                        await pipe.execute()
                         item["_skip"] = True
                         continue
 
-                    # FIX #3: retries default 0 PEHLE — UnboundLocalError prevent
-                    retries = 0
+                    # --------------------------------------------------
+                    # 4. Process each lead
+                    # --------------------------------------------------
                     try:
-                        retries = int(await self.client.hget(RETRY_COUNT_KEY, lead_id) or 0)
+                        retries    = int(await self.client.hget(RETRY_COUNT_KEY, lead_id) or 0)
                         start_time = time.time()
 
-                        # Run Automation
                         success, err = await engine.process_lead(lead_email)
                         elapsed = (time.time() - start_time) * 1000
 
@@ -372,18 +455,20 @@ class WorkerTask:
                             "error": err if not success else None,
                             "processing_time_ms": elapsed,
                             "worker_id": self.task_id,
-                            "row_number": lead_data.get("row_number")
+                            "row_number": lead_data.get("row_number"),
+                            "_retries": retries,
                         })
 
                         if success:
                             self.leads_done += 1
-                            logger.info(f"[+] Successfully sent lead to: {lead_email}")
+                            logger.info(f"[+] Sent lead: {lead_email} via {email}")
                         else:
                             self.leads_failed += 1
                             logger.warning(f"[-] Failed lead: {lead_email}. Reason: {err}")
 
                         if err == "rate_limit":
-                            logger.warning("Rate limit hit, stopping batch")
+                            logger.warning("Rate limit hit — stopping batch early.")
+                            # Un-claim remaining batch items that weren't processed
                             break
 
                     except Exception as e:
@@ -397,12 +482,15 @@ class WorkerTask:
                                 "success": False,
                                 "error": str(e),
                                 "worker_id": self.task_id,
-                                "row_number": lead_data.get("row_number")
+                                "row_number": lead_data.get("row_number"),
+                                "_retries": retries,
                             })
                         else:
+                            # Requeue and un-claim
                             pipe = self.client.pipeline()
                             pipe.rpush(LEAD_QUEUE, raw_data)
                             pipe.lrem(PROCESSING_QUEUE, -1, raw_data)
+                            pipe.srem(f"used:{email}", lead_email)
                             await pipe.execute()
                             item["_skip"] = True
 
@@ -411,79 +499,94 @@ class WorkerTask:
             finally:
                 await engine.stop()
 
-            # 4. Handle Account Routing (Cooldown / Done / Pending Refresh)
+            # ----------------------------------------------------------
+            # 5. Account routing after batch
+            # ----------------------------------------------------------
             if not login_success:
                 await self.client.sadd("account_pool:pending_refresh", email)
-                logger.info(f"Account {email} login failed. Moved to pending_refresh.")
+                logger.info(f"Account {email} login failed → pending_refresh.")
             else:
-                # FIX #6: Count only actually processed leads (not skipped)
-                actual_processed = len([b for b in batch if not b.get("_skip")])
-                leads_sent = await self.client.hincrby(acc_key, "leads_sent_today", actual_processed)
+                # Count only leads that were actually attempted (not skipped/requeued)
+                actually_sent = sum(
+                    1 for r in results
+                    if not r.get("_skip") and r["success"]
+                )
+                leads_sent = await self.client.hincrby(acc_key, "leads_sent_today", actually_sent)
 
                 if leads_sent >= MAX_DAILY_LEADS:
                     await self.client.sadd(ACCOUNT_POOL_DONE, email)
-                    logger.info(f"Account {email} hit daily limit ({leads_sent}/{MAX_DAILY_LEADS}). Moved to Done.")
+                    logger.info(f"Account {email} hit daily limit ({leads_sent}/{MAX_DAILY_LEADS}) → Done.")
                 else:
                     now = time.time()
                     await self.client.hset(acc_key, "cooldown_until", now + COOLDOWN_SECONDS)
                     await self.client.sadd(ACCOUNT_POOL_COOLDOWN, email)
-                    logger.info(f"Account {email} on Cooldown ({leads_sent}/{MAX_DAILY_LEADS}). 1 hour rest.")
+                    logger.info(
+                        f"Account {email} batch done ({leads_sent}/{MAX_DAILY_LEADS}) "
+                        f"→ Cooldown 1hr. Next batch in ~1 hour."
+                    )
 
-            final_batch = [b for b in batch if not b.get("_skip")]
-            valid_ids = {b["data"].get("id") for b in final_batch}
-            final_results = [r for r in results if r["lead_id"] in valid_ids]
+            # ----------------------------------------------------------
+            # 6. Push results to result_queue and clean processing_queue
+            # ----------------------------------------------------------
+            final_batch   = [b for b in batch  if not b.get("_skip")]
+            valid_ids     = {b["data"].get("id") for b in final_batch}
+            final_results = [r for r in results if r.get("lead_id") in valid_ids and not r.get("_skip")]
 
-            await self._cleanup_and_push_results(final_batch, final_results)
+            await self._cleanup_and_push_results(final_batch, final_results, email)
 
 
+# Global list of all WorkerTask instances on this machine
 _all_worker_tasks: list["WorkerTask"] = []
 
 
 async def machine_heartbeat_loop():
-    """Send ONE heartbeat per machine every 5s."""
+    """Sends ONE heartbeat key for the whole machine every 5s."""
     client = redis.Redis(connection_pool=get_redis_pool())
-    key = f"worker:machine:{MACHINE_ID}"
+    key    = f"worker:machine:{MACHINE_ID}"
+
     while not _shutdown_requested:
         async with _stats_lock:
-            done = _machine_stats["total_done"]
-            failed = _machine_stats["total_failed"]
+            done    = _machine_stats["total_done"]
+            failed  = _machine_stats["total_failed"]
             threads = dict(_machine_stats["threads"])
 
-        total = done + failed
+        total        = done + failed
         success_rate = round((done / total * 100), 1) if total > 0 else 0
-        busy_count = sum(1 for a in threads.values() if a not in ("Idle", "Starting"))
+        busy_count   = sum(1 for a in threads.values() if a not in ("Idle", "Starting"))
 
         payload = {
-            "id": MACHINE_ID,
-            "status": "busy" if busy_count > 0 else "online",
+            "id":             MACHINE_ID,
+            "status":         "busy" if busy_count > 0 else "online",
             "active_threads": busy_count,
-            "max_threads": WORKER_TASKS,
-            "leads_done": done,
-            "leads_failed": failed,
-            "success_rate": success_rate,
+            "max_threads":    WORKER_TASKS,
+            "leads_done":     done,
+            "leads_failed":   failed,
+            "success_rate":   success_rate,
             "current_action": next(
                 (a for a in threads.values() if a not in ("Idle", "Starting")),
                 "Idle"
             ),
-            "version": "2.1.0",
-            "last_seen": time.time()
+            "version":   "3.0.0",
+            "last_seen": time.time(),
         }
         await client.setex(key, 30, json.dumps(payload))
         await asyncio.sleep(5)
 
 
 async def main():
-    global _stats_lock
-
-    # FIX #13: asyncio.Lock() event loop ke andar banao, module level par nahi
-    _stats_lock = asyncio.Lock()
-
     logger.info("=" * 50)
     logger.info(f"Starting Async Playwright Worker [{MACHINE_ID}]. Tasks: {WORKER_TASKS}")
     logger.info("=" * 50)
 
-    # FIX #12: Redis connection check at startup
-    await check_redis_connection()
+    # Set SENT_LEADS_SET to expire at end of day so it auto-resets
+    # (The account_manager midnight reset also handles this via leads_sent_today)
+    client = redis.Redis(connection_pool=get_redis_pool())
+    import datetime
+    now = datetime.datetime.now(datetime.timezone.utc)
+    tomorrow = now.replace(hour=0, minute=0, second=0, microsecond=0) + datetime.timedelta(days=1)
+    ttl = int((tomorrow - now).total_seconds())
+    await client.expire(SENT_LEADS_SET, ttl)
+    logger.info(f"SENT_LEADS_SET TTL set to {ttl}s (expires at UTC midnight).")
 
     for i in range(WORKER_TASKS):
         t = WorkerTask(f"worker-{uuid.uuid4().hex[:6]}")