deploy fix, removed celery and reids

auth.py

@@ -53,7 +53,7 @@ def get_dev_user() -> Optional[User]:
         
     try:
         # Use DEV_USER_ID if provided, otherwise use a default dev ID
-        dev_user_id = os.getenv("DEV_USER_ID", "cc310cdf-83af-48e4-a341-d6c2f3b5462d")
+        dev_user_id = os.getenv("DEV_USER_ID", "dev-user-id-change-this")
         
         print(f"✅ Creating dev user: {DEV_USER_EMAIL} (ID: {dev_user_id})")
         return User(

requirements.txt

@@ -12,6 +12,4 @@ stripe>=5.0.0
 python-multipart>=0.0.20
 requests>=2.32.3
 PyJWT>=2.10.1
-celery>=5.3.0
-redis>=5.0.0
 sse-starlette>=1.3.2

routers/user_models.py

@@ -9,8 +9,7 @@ import asyncio
 from pydantic import BaseModel
 from utils.worker import poll_meshy_task
 import json
-import redis
-import redis.asyncio as aioredis
+from utils.pubsub import pubsub as inmem_pubsub
 
 router = APIRouter(
     prefix="/user/models",
@@ -109,42 +108,31 @@ async def progress_update(model_id: str, current_user: User = Depends(get_curren
 async def sse_progress(model_id: str, request: Request, current_user: User = Depends(get_current_active_user)):
     """
     Server-Sent Events endpoint that streams realtime progress updates for a
-    specific `model_id`. The Celery task publishes JSON messages to the
-    Redis channel "tasks.progress". We listen on that pub/sub channel and
-    forward only the events that match the requested model id.
+    specific `model_id`. The background polling task publishes JSON messages to the
+    internal in-memory pub/sub bus. We listen on that bus and forward only
+    the events that match the requested task id.
     """
 
-    # Establish async Redis connection
-    redis_url = os.getenv("REDIS_URL", "redis://localhost/0")
-    redis_conn = aioredis.from_url(redis_url, decode_responses=True)
-
     async def event_generator():
-        pubsub = redis_conn.pubsub()
-        await pubsub.subscribe("tasks.progress")
+        # Subscribe to in-memory pub/sub for this task
+        queue = await inmem_pubsub.subscribe(model_id)
         try:
             while True:
-                # Client has gone away – break the loop to finish response
                 if await request.is_disconnected():
                     break
 
-                message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=1)
-                if message and message.get("type") == "message":
-                    try:
-                        data = json.loads(message["data"])
-                    except json.JSONDecodeError:
-                        data = None
+                try:
+                    data = await asyncio.wait_for(queue.get(), timeout=1.0)
+                except asyncio.TimeoutError:
+                    data = None
 
-                    # Forward only messages for the requested task
-                    if data and data.get("taskId") == model_id:
-                        # Yield SSE-compatible dictionary. `sse-starlette`
-                        # handles conversion to the correct wire format.
-                        yield {"data": json.dumps(data)}
+                if data:
+                    yield {"data": json.dumps(data)}
 
-                # Give control back to the event loop
+                # Small sleep to relinquish control
                 await asyncio.sleep(0.1)
         finally:
-            await pubsub.unsubscribe("tasks.progress")
-            await pubsub.close()
+            await inmem_pubsub.unsubscribe(model_id, queue)
 
     return EventSourceResponse(event_generator())
 
@@ -167,9 +155,9 @@ async def _check_and_decrement_credits(user_id: str):
 
 
 @router.post("/text-to-3d")
-async def text_to_3d(prompt: TextPrompt, current_user: User = Depends(get_current_active_user)):
+async def text_to_3d(prompt: TextPrompt, background_tasks: BackgroundTasks, current_user: User = Depends(get_current_active_user)):
     """Reframe user prompt via OpenAI, create a Meshy Text-to-3D task, and kick
-    off a Celery poller that streams progress via SSE."""
+    off a FastAPI background poller that streams progress via SSE."""
 
     # Credit check and decrement
     await _check_and_decrement_credits(current_user.id)
@@ -281,23 +269,20 @@ async def text_to_3d(prompt: TextPrompt, current_user: User = Depends(get_curren
         except Exception as ex:
             logging.warning(f"Failed to update DB with Meshy taskId: {ex}")
 
-    # 6. Kick off Celery poller + send initial SSE event
+    # 6. Kick off background poller + send initial SSE event
     if generated_model_id:
-        # publish immediate queued event
+        # publish immediate queued event via in-memory bus
         try:
-            redis.Redis(host="localhost", decode_responses=True).publish(
-                "tasks.progress",
-                json.dumps({
-                    "taskId": str(generated_model_id),
-                    "progress": 0,
-                    "status": "queued",
-                }),
-            )
+            await inmem_pubsub.publish(str(generated_model_id), {
+                "taskId": str(generated_model_id),
+                "progress": 0,
+                "status": "queued",
+            })
         except Exception as ex:
             logging.warning(f"Failed to publish initial SSE event: {ex}")
 
-        # Fire background polling task
-        poll_meshy_task.delay(generated_model_id, meshy_task_id)
+        # Fire background polling task using FastAPI BackgroundTasks
+        background_tasks.add_task(poll_meshy_task, generated_model_id, meshy_task_id)
 
     return {
         "generated_model_id": generated_model_id,

utils/pubsub.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import asyncio
+from collections import defaultdict
+from typing import Dict, List, Any
+
+
+class InMemoryPubSub:
+    """A minimal in-process pub/sub helper.
+
+    Subscribers are simply *asyncio.Queue* instances.  Each task_id can have
+    many subscribers (e.g. multiple browser tabs connected to the same SSE
+    stream).  The publisher pushes a message **dict** (already JSON-serialisable)
+    and every subscriber receives it.
+
+    NOTE: This works only inside a single Python process.  If you run several
+    Uvicorn/Gunicorn workers the events won't be shared between them.  Use a
+    proper external broker (Redis, NATS, …) for multi-process deployments.
+    """
+
+    def __init__(self) -> None:
+        self._subscribers: Dict[str, List[asyncio.Queue]] = defaultdict(list)
+        self._lock = asyncio.Lock()  # protect subscriber mutations
+
+    async def publish(self, task_id: str, message: Any) -> None:
+        """Broadcast *message* to every subscriber listening to *task_id*."""
+        # Work on a snapshot to avoid race conditions if someone unsubscribes
+        async with self._lock:
+            queues = list(self._subscribers.get(task_id, []))
+        for queue in queues:
+            await queue.put(message)
+
+    async def subscribe(self, task_id: str) -> asyncio.Queue:
+        """Return a brand-new *asyncio.Queue* subscribed to *task_id*."""
+        queue: asyncio.Queue = asyncio.Queue()
+        async with self._lock:
+            self._subscribers[task_id].append(queue)
+        return queue
+
+    async def unsubscribe(self, task_id: str, queue: asyncio.Queue) -> None:
+        """Remove *queue* from *task_id* subscriptions and cleanup."""
+        async with self._lock:
+            if task_id in self._subscribers and queue in self._subscribers[task_id]:
+                self._subscribers[task_id].remove(queue)
+                if not self._subscribers[task_id]:
+                    del self._subscribers[task_id]
+
+
+# Global singleton used across the code-base
+pubsub = InMemoryPubSub()
+
+__all__ = ["pubsub", "InMemoryPubSub"] 

utils/worker.py

@@ -1,33 +1,14 @@
-import os, json, time, random
-from celery import Celery
-import redis
+import os, asyncio, logging
 import httpx
 from auth import supabase  # re-use existing Supabase client
-
-R = redis.Redis(host="localhost", decode_responses=True)
-celery = Celery("tasks", broker=os.getenv("REDIS_URL", "redis://localhost/0"))
-
-@celery.task(bind=True)
-def external_job(self, task_id: str, payload: dict):
-    """
-    Poll a fictitious external API until done, emitting progress every ~2 s.
-    """
-    for p in range(0, 101, 10):
-        # --- call the real external API here ----------------------------
-        time.sleep(random.uniform(1.5, 2.5))               # <- placeholder
-        # ----------------------------------------------------------------
-        status = "running" if p < 100 else "completed"
-        event = dict(taskId=task_id, progress=p, status=status)
-        R.publish("tasks.progress", json.dumps(event))      # push to bus
-    return {"result": "OK"}
+from utils.pubsub import pubsub as inmem_pubsub
 
 # ---------------------------------------------------------------------------
-# Real Meshy Text-to-3D polling task
+# Real Meshy Text-to-3D polling task – executed as FastAPI BackgroundTask
 # ---------------------------------------------------------------------------
 
-@celery.task(bind=True, max_retries=3, default_retry_delay=30)
-def poll_meshy_task(self, generated_model_id: int, meshy_task_id: str):
-    """Poll Meshy API every few seconds, publish progress to Redis and update DB.
+async def poll_meshy_task(generated_model_id: int, meshy_task_id: str):
+    """Poll Meshy API every few seconds, publish progress to in-memory pub/sub and update DB.
 
     Parameters
     ----------
@@ -40,53 +21,52 @@ def poll_meshy_task(self, generated_model_id: int, meshy_task_id: str):
 
     meshy_api_key = os.getenv("MESHY_API_KEY")
     if not meshy_api_key:
-        raise RuntimeError("MESHY_API_KEY env-var missing – cannot poll Meshy API")
+        logging.error("MESHY_API_KEY env-var missing – cannot poll Meshy API")
+        return
 
     headers = {"Authorization": f"Bearer {meshy_api_key}"}
 
-    with httpx.Client(timeout=30.0) as client:
-        last_progress = -1  # to avoid spamming identical values
+    last_progress = -1  # to avoid spamming identical values
+
+    async with httpx.AsyncClient(timeout=30.0) as client:
         while True:
             try:
-                resp = client.get(
+                resp = await client.get(
                     f"https://api.meshy.ai/openapi/v2/text-to-3d/{meshy_task_id}",
                     headers=headers,
                 )
             except Exception as exc:
-                # network hiccup – retry a few times
-                self.retry(exc=exc)
+                logging.warning(f"Meshy API request failed: {exc}; retrying in 30s")
+                await asyncio.sleep(30)
+                continue
 
             if resp.status_code != 200:
-                # Non-200 from Meshy; treat as terminal failure and notify.
-                status = "FAILED"
+                # Meshy returned an error – notify and abort.
                 data = {
                     "taskId": str(generated_model_id),
-                    "progress": last_progress if last_progress >= 0 else 0,
-                    "status": status,
+                    "progress": max(last_progress, 0),
+                    "status": "failed",
                     "error": f"Meshy API error {resp.status_code}",
                 }
-                R.publish("tasks.progress", json.dumps(data))
+                await inmem_pubsub.publish(str(generated_model_id), data)
                 break
 
             payload = resp.json()
             status = payload.get("status", "UNKNOWN")
             progress = payload.get("progress", 0)
 
-            # Publish only if progress changed or when completed/failed
+            # Publish progress updates (avoid duplicates)
             if progress != last_progress or status in ("SUCCEEDED", "FAILED"):
-                R.publish(
-                    "tasks.progress",
-                    json.dumps({
-                        "taskId": str(generated_model_id),
-                        "progress": progress,
-                        "status": status.lower(),  # 'succeeded' -> 'succeeded'
-                        "model_urls": payload.get("model_urls"),
-                        "thumbnail_url": payload.get("thumbnail_url"),
-                    }),
-                )
+                await inmem_pubsub.publish(str(generated_model_id), {
+                    "taskId": str(generated_model_id),
+                    "progress": progress,
+                    "status": status.lower(),
+                    "model_urls": payload.get("model_urls"),
+                    "thumbnail_url": payload.get("thumbnail_url"),
+                })
                 last_progress = progress
 
-            # If finished, update the DB and exit loop
+            # Handle terminal states
             if status == "SUCCEEDED":
                 try:
                     supabase.from_("Generated_Models").update({
@@ -94,10 +74,10 @@ def poll_meshy_task(self, generated_model_id: int, meshy_task_id: str):
                         "status": "COMPLETED",
                         "updated_at": "now()",
                     }).eq("generated_model_id", generated_model_id).execute()
-                except Exception:
-                    # Non-critical – continue anyway
-                    pass
+                except Exception as exc:
+                    logging.warning(f"Failed to update DB on success: {exc}")
                 break
+
             if status == "FAILED":
                 try:
                     supabase.from_("Generated_Models").update({
@@ -105,9 +85,12 @@ def poll_meshy_task(self, generated_model_id: int, meshy_task_id: str):
                         "status": "FAILED",
                         "updated_at": "now()",
                     }).eq("generated_model_id", generated_model_id).execute()
-                except Exception:
-                    pass
+                except Exception as exc:
+                    logging.warning(f"Failed to update DB on failure: {exc}")
                 break
 
-            # Keep polling every 5 seconds
-            time.sleep(5)
+            # Sleep between polls
+            await asyncio.sleep(5)
+
+# Provide explicit __all__ for clarity
+__all__ = ["poll_meshy_task"]