Implement product catalogue and order management features#67
Conversation
- Added product model and schema for CRUD operations. - Integrated product routes into the API. - Developed search functionality for products within the agent system. - Created frontend components for product listing, adding, editing, and bulk uploading. - Implemented notifications for user actions related to products. - Added CSV import functionality for bulk product uploads. - Updated dashboard layout to include a link to the product catalogue.
- Implemented API functions for listing orders, retrieving a single order, and updating order status in api.ts. - Defined order-related types including Order, OrderItem, and OrdersPage in types.ts.
|
🤖 Hi @stenwire, I've received your request, and I'm working on it now! You can track my progress in the logs for more details. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a comprehensive WhatsApp broadcast campaign system, a product catalogue, and order management capabilities, alongside pagination and search enhancements across multiple API endpoints and frontend views. The code review identifies several critical issues, including a runtime NameError due to missing imports in the WhatsApp API, database portability concerns from raw PostgreSQL queries in the campaign worker, and an anti-pattern in database session retrieval. Additionally, the feedback recommends using Decimal instead of float for monetary values to prevent precision errors and adding a transaction rollback on bulk upload failures.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| from app.models.whatsapp_broadcast import ( | ||
| CampaignMessageStatus, | ||
| WhatsAppCampaign, | ||
| WhatsAppCampaignMessage, | ||
| ) |
There was a problem hiding this comment.
The _process_status_updates function uses datetime and timezone (e.g., datetime.now(timezone.utc)), but these are not imported in this file. This will raise a NameError at runtime when a status update webhook is received. Please import datetime and timezone from the datetime module.
| from app.models.whatsapp_broadcast import ( | |
| CampaignMessageStatus, | |
| WhatsAppCampaign, | |
| WhatsAppCampaignMessage, | |
| ) | |
| from datetime import datetime, timezone | |
| from app.models.whatsapp_broadcast import ( | |
| CampaignMessageStatus, | |
| WhatsAppCampaign, | |
| WhatsAppCampaignMessage, | |
| ) |
| result = db.execute( | ||
| text( | ||
| """ | ||
| SELECT id FROM whatsapp_campaigns | ||
| WHERE status = :scheduled | ||
| AND scheduled_at <= :now | ||
| ORDER BY scheduled_at ASC | ||
| LIMIT :limit | ||
| FOR UPDATE SKIP LOCKED | ||
| """ | ||
| ), | ||
| { | ||
| "scheduled": CampaignStatus.SCHEDULED.value, | ||
| "now": now, | ||
| "limit": CLAIM_LIMIT, | ||
| }, | ||
| ) | ||
| ids = [row[0] for row in result] | ||
| if not ids: | ||
| db.commit() | ||
| return [] | ||
|
|
||
| # Mark as SENDING inside the same transaction while the lock is held. | ||
| db.execute( | ||
| text( | ||
| """ | ||
| UPDATE whatsapp_campaigns | ||
| SET status = :sending, started_at = :now, updated_at = :now | ||
| WHERE id = ANY(:ids) | ||
| """ | ||
| ), | ||
| { | ||
| "sending": CampaignStatus.SENDING.value, | ||
| "now": now, | ||
| "ids": ids, | ||
| }, | ||
| ) | ||
| db.commit() |
There was a problem hiding this comment.
The raw SQL query uses PostgreSQL-specific syntax WHERE id = ANY(:ids) and FOR UPDATE SKIP LOCKED which will fail when running unit tests or local development against SQLite. To ensure database portability and maintainability, rewrite these queries using SQLAlchemy's database-agnostic ORM methods.
# Use SQLAlchemy ORM for database-agnostic query execution (works on both PostgreSQL and SQLite)
result = (
db.query(WhatsAppCampaign.id)
.filter(
WhatsAppCampaign.status == CampaignStatus.SCHEDULED.value,
WhatsAppCampaign.scheduled_at <= now,
)
.order_by(WhatsAppCampaign.scheduled_at.asc())
.limit(CLAIM_LIMIT)
.with_for_update(skip_locked=True)
.all()
)
ids = [row[0] for row in result]
if not ids:
db.commit()
return []
# Mark as SENDING inside the same transaction while the lock is held.
db.query(WhatsAppCampaign).filter(WhatsAppCampaign.id.in_(ids)).update(
{
WhatsAppCampaign.status: CampaignStatus.SENDING.value,
WhatsAppCampaign.started_at: now,
WhatsAppCampaign.updated_at: now,
},
synchronize_session=False,
)
db.commit()| # Delivery / read status callbacks from outbound campaigns. | ||
| statuses = value.get("statuses") | ||
| if statuses: | ||
| db = next(get_db()) |
There was a problem hiding this comment.
Using next(get_db()) to manually retrieve a database session is an anti-pattern and can lead to generator lifecycle issues or unclosed sessions. Since this is a webhook handler running outside of FastAPI's dependency injection context, it is safer and cleaner to use SessionLocal() directly with a try...finally block.
from app.db.session import SessionLocal
db = SessionLocal()| price_str = row.get('price', '0').replace(',', '') | ||
| price = float(price_str) if price_str else 0.0 |
There was a problem hiding this comment.
Parsing currency/price values as float can introduce floating-point precision and rounding errors. Since the database column Product.price is defined as Numeric(10, 2) and the Pydantic schema expects a Decimal, use Python's Decimal type to parse and store monetary values accurately.
| price_str = row.get('price', '0').replace(',', '') | |
| price = float(price_str) if price_str else 0.0 | |
| from decimal import Decimal | |
| price_str = row.get('price', '0').replace(',', '') | |
| price = Decimal(price_str) if price_str else Decimal("0.00") |
| except Exception as e: | ||
| raise HTTPException(status_code=400, detail=f"Failed to process CSV file: {str(e)}") |
There was a problem hiding this comment.
If an exception occurs during database operations or db.commit(), the transaction is left open and un-rolled-back, which can leave the database session in an invalid state. Ensure db.rollback() is called in the except block before raising the HTTPException.
| except Exception as e: | |
| raise HTTPException(status_code=400, detail=f"Failed to process CSV file: {str(e)}") | |
| except Exception as e: | |
| db.rollback() | |
| raise HTTPException(status_code=400, detail=f"Failed to process CSV file: {str(e)}") |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
This is a substantial pull request that introduces significant new functionality for product catalogues, order management, and WhatsApp broadcasting. The backend implementation is generally robust and well-structured. The new database models, APIs, and background worker are well-designed.
I've added a few inline comments with suggestions for performance improvements and to enhance code clarity and robustness, but these are minor.
🔍 General Feedback
- Good Job: The separation of concerns between services, API routers, and the new background worker is clean. The use of
FOR UPDATE SKIP LOCKEDfor the campaign worker is a great choice for building a scalable job queue. - Testing: Given the size and complexity of this feature, it would be beneficial to add more integration tests, especially for the CSV import and the campaign execution logic, to cover various edge cases.
- Frontend: The new frontend pages look good and seem to correctly integrate with the new APIs. The addition of pagination to existing pages is also a welcome improvement.
Overall, this is a great contribution that adds a lot of value.
| else: | ||
| product = Product( |
There was a problem hiding this comment.
| else: | |
| product = Product( | |
| price_str = row.get('price', '0').replace(',', '') | |
| price = float(price_str) if price_str else 0.0 | |
| stock_str = row.get('stock_quantity', '0').strip() | |
| stock = int(stock_str) if stock_str else 0 | |
| is_active = row.get('is_active', 'true').lower() == 'true' |
| from_phone, | ||
| ai_text, | ||
| ) | ||
|
|
There was a problem hiding this comment.
To improve performance, you can move the db.commit() outside the loop and cache campaign objects.
| def _process_status_updates(db: Session, statuses: list[dict]) -> None: | |
| """Update campaign messages from Meta `statuses[]` webhook entries.""" | |
| campaign_cache = {} | |
| for entry in statuses: | |
| meta_id = entry.get("id") | |
| raw_status = (entry.get("status") or "").lower() | |
| if not meta_id or raw_status not in _STATUS_TO_ENUM: | |
| continue | |
| new_status = _STATUS_TO_ENUM[raw_status] | |
| msg = ( | |
| db.query(WhatsAppCampaignMessage) | |
| .filter(WhatsAppCampaignMessage.meta_message_id == meta_id) | |
| .first() | |
| ) | |
| if not msg: | |
| continue | |
| # Idempotency / out-of-order protection: only move forward. | |
| current_rank = _STATUS_RANK.get(msg.status, 0) | |
| new_rank = _STATUS_RANK[new_status.value] | |
| if new_rank <= current_rank and new_status != CampaignMessageStatus.FAILED: | |
| continue | |
| previous_status = msg.status | |
| msg.status = new_status.value | |
| now = datetime.now(timezone.utc) | |
| if new_status == CampaignMessageStatus.DELIVERED and not msg.delivered_at: | |
| msg.delivered_at = now | |
| elif new_status == CampaignMessageStatus.READ and not msg.read_at: | |
| msg.read_at = now | |
| elif new_status == CampaignMessageStatus.FAILED: | |
| errors = entry.get("errors") or [] | |
| if errors: | |
| err = errors[0] | |
| msg.error_code = str(err.get("code", ""))[:100] | |
| msg.error_message = str(err.get("title") or err.get("message") or "")[:500] | |
| # Aggregate into campaign counts (only on first transition into each state). | |
| campaign = campaign_cache.get(msg.campaign_id) | |
| if campaign is None: | |
| campaign = ( | |
| db.query(WhatsAppCampaign) | |
| .filter(WhatsAppCampaign.id == msg.campaign_id) | |
| .first() | |
| ) | |
| campaign_cache[msg.campaign_id] = campaign | |
| if campaign and previous_status != new_status.value: | |
| if new_status == CampaignMessageStatus.DELIVERED: | |
| campaign.delivered_count += 1 | |
| elif new_status == CampaignMessageStatus.READ: | |
| campaign.read_count += 1 | |
| elif new_status == CampaignMessageStatus.FAILED: | |
| campaign.failed_count += 1 | |
| if previous_status == CampaignMessageStatus.SENT.value: | |
| campaign.sent_count = max(0, campaign.sent_count - 1) | |
| db.commit() |
| f"ROUTING RULES — you MUST delegate every request, NEVER respond directly:\n" | ||
| f"- Greetings (hello, hi, hey, good morning, etc.): ALWAYS delegate to 'greeting_agent'\n" | ||
| f"- Farewells (bye, goodbye, see you, thanks and bye, etc.): ALWAYS delegate to 'farewell_agent'\n" | ||
| f"- Products, prices, stock, purchasing, or order questions: ALWAYS delegate to 'sales_agent'\n" |
There was a problem hiding this comment.
The rag_agent is already designed to handle out-of-scope questions gracefully. It's better to make the rag_agent the default fallback for any request that doesn't fit a more specific category. This makes the chief agent's routing logic simpler and more robust.
| f"ROUTING RULES — you MUST delegate every request, NEVER respond directly:\n" | |
| f"- Greetings (hello, hi, hey, good morning, etc.): ALWAYS delegate to 'greeting_agent'\n" | |
| f"- Farewells (bye, goodbye, see you, thanks and bye, etc.): ALWAYS delegate to 'farewell_agent'\n" | |
| f"- Products, prices, stock, purchasing, or order questions: ALWAYS delegate to 'sales_agent'\n" | |
| f"ROUTING RULES — you MUST delegate every request, NEVER respond directly:\n" | |
| f"- Greetings (hello, hi, hey, good morning, etc.): ALWAYS delegate to 'greeting_agent'\n" | |
| f"- Farewells (bye, goodbye, see you, thanks and bye, etc.): ALWAYS delegate to 'farewell_agent'\n" | |
| f"- Products, prices, stock, purchasing, or order questions: ALWAYS delegate to 'sales_agent'\n" | |
| f"- General info, FAQ, how-to, support, or any other questions about {business_name}: ALWAYS delegate to 'rag_agent'\n" | |
| f"- User frustrated, requests a human, or asks to speak to someone: ALWAYS delegate to 'escalation_agent'\n" | |
| f"- Anything outside {business_name} topics: ALWAYS delegate to 'rag_agent'.\n\n" | |
| f"IMPORTANT: You have NO tools of your own. You MUST delegate to the appropriate agent for every request. Do NOT compose a reply yourself.\n\n" |
|
|
||
| user_id = tool_context.state.get("user_id") | ||
| if not user_id: |
There was a problem hiding this comment.
| user_id = tool_context.state.get("user_id") | |
| if not user_id: | |
| import json | |
| from decimal import Decimal | |
| import asyncio | |
| import threading | |
| from app.core.subscription import TIER_LIMITS | |
| from app.core.config import settings | |
| from app.services.email_service import EmailSchema |
(This would be at the top of the file)
|
|
||
|
|
||
| def import_csv(db: Session, business_id: str, file_bytes: bytes) -> CsvImportResult: | ||
| """Import contacts from a CSV with headers: phone, name (optional), tags (optional, comma-sep).""" |
There was a problem hiding this comment.
You can significantly improve performance by fetching all potentially existing contacts in a single query before the loop.
| """Import contacts from a CSV with headers: phone, name (optional), tags (optional, comma-sep).""" | |
| def import_csv(db: Session, business_id: str, file_bytes: bytes) -> CsvImportResult: | |
| """Import contacts from a CSV with headers: phone, name (optional), tags (optional, comma-sep).""" | |
| result = CsvImportResult() | |
| try: | |
| text = file_bytes.decode("utf-8-sig") | |
| except UnicodeDecodeError: | |
| result.errors.append("File is not valid UTF-8") | |
| return result | |
| reader = csv.DictReader(io.StringIO(text)) | |
| if not reader.fieldnames or "phone" not in [h.lower() for h in reader.fieldnames]: | |
| result.errors.append("CSV must have a 'phone' header column") | |
| return result | |
| rows = list(reader) | |
| phones_in_csv = { | |
| normalize_phone(row.get(key) or "") | |
| for row in rows | |
| for key in row | |
| if key and key.lower() == "phone" and row[key] | |
| } | |
| phones_in_csv.discard(None) | |
| existing_contacts = { | |
| c.phone_e164: c | |
| for c in db.query(WhatsAppContact).filter( | |
| WhatsAppContact.business_id == business_id, | |
| WhatsAppContact.phone_e164.in_(phones_in_csv), | |
| ) | |
| } | |
| # Normalize headers to lowercase for lookup | |
| for row_num, row in enumerate(rows, start=2): | |
| row_lower = {(k or "").lower(): v for k, v in row.items()} | |
| raw_phone = (row_lower.get("phone") or "").strip() | |
| if not raw_phone: | |
| result.skipped += 1 | |
| continue | |
| normalized = normalize_phone(raw_phone) | |
| if not normalized: | |
| result.errors.append(f"Row {row_num}: invalid phone '{raw_phone}'") | |
| result.skipped += 1 | |
| continue | |
| name = (row_lower.get("name") or "").strip() or None | |
| tags_raw = (row_lower.get("tags") or "").strip() | |
| tags = [t.strip() for t in tags_raw.split(",") if t.strip()] if tags_raw else None | |
| if normalized in existing_contacts: | |
| existing = existing_contacts[normalized] | |
| if name and not existing.name: | |
| existing.name = name | |
| if tags: | |
| existing.tags = tags | |
| result.skipped += 1 | |
| continue | |
| contact = WhatsAppContact( | |
| business_id=business_id, | |
| phone_e164=normalized, | |
| name=name, | |
| tags=tags, | |
| source=ContactSource.CSV.value, | |
| opted_in=True, | |
| ) | |
| db.add(contact) | |
| result.imported += 1 | |
| db.commit() | |
| return result |
1 participant
Description
Type of Change
Checklist
make lint-be/make lint-fe)make test-be/make test-fe)Screenshots / Demo
Related Issues