support unregistered drivers from group messages and warm new user onboarding

app/api/routes.py

@@ -68,7 +68,14 @@ async def receive_whatsapp_webhook(
     payload: dict[str, Any] = await request.json()
     messages = parse_inbound_messages(payload)
     for inbound in messages:
-        background_tasks.add_task(container.conversation.handle_inbound_message, inbound)
+        if inbound.is_group:
+            background_tasks.add_task(
+                container.group_message.handle_group_message, inbound
+            )
+        else:
+            background_tasks.add_task(
+                container.conversation.handle_inbound_message, inbound
+            )
 
     return WebhookAcceptedResponse(messages=len(messages))
 
@@ -90,9 +97,12 @@ async def receive_whatsapp_webhook_debug(
     messages = parse_inbound_messages(payload)
     replies: list[str] = []
     for inbound in messages:
-        reply = await container.conversation.handle_inbound_message(inbound)
-        if reply is not None:
-            replies.append(reply)
+        if inbound.is_group:
+            await container.group_message.handle_group_message(inbound)
+        else:
+            reply = await container.conversation.handle_inbound_message(inbound)
+            if reply is not None:
+                replies.append(reply)
 
     return WebhookDebugResponse(messages=len(messages), replies=replies)
 

app/database/supabase.py

@@ -39,6 +39,7 @@ class SupabaseRepository:
         name: str | None = None,
         preferred_language: str | None = None,
         phone_number: str | None = None,
+        registered: bool = True,
     ) -> dict[str, Any]:
         if phone_number:
             phone_number = phone_number.split("@")[0]
@@ -47,6 +48,7 @@ class SupabaseRepository:
             "name": name,
             "preferred_language": preferred_language,
             "phone_number": phone_number,
+            "registered": registered,
         }
         payload = {key: value for key, value in payload.items() if value is not None}
         response = await (
@@ -57,6 +59,110 @@ class SupabaseRepository:
         data = _response_data(response)
         return data[0] if isinstance(data, list) else data
 
+    async def get_customer_by_phone_number(
+        self,
+        phone_number: str,
+    ) -> dict[str, Any] | None:
+        response = await (
+            self.client.table("customers")
+            .select("*")
+            .eq("phone_number", phone_number)
+            .maybe_single()
+            .execute()
+        )
+        return _response_data(response)
+
+    async def create_unregistered_driver_entities(
+        self,
+        *,
+        phone_number: str,
+        driver_name: str | None,
+        car_type: str | None,
+        departure: str,
+        destination: str,
+        departure_date: Any,
+        departure_time: str,
+        available_seats: int,
+        total_seats: int,
+        price: float,
+    ) -> dict[str, Any]:
+        customer = await self.upsert_customer(
+            remote_jid=phone_number,
+            name=driver_name,
+            phone_number=phone_number,
+            registered=False,
+        )
+        driver = await self.create_driver(customer_id=str(customer["id"]))
+        car = await self.create_driver_car(
+            driver_id=str(driver["id"]),
+            car_type=car_type or "غير معروف",
+            seat_count=total_seats,
+        )
+        trip = await self.create_driver_trip(
+            driver_id=str(driver["id"]),
+            car_id=str(car["id"]),
+            departure=departure,
+            destination=destination,
+            departure_date=departure_date,
+            departure_time=departure_time,
+            available_seats=available_seats,
+            total_seats=total_seats,
+            price=price,
+        )
+        return trip
+
+    async def create_unregistered_driver_trip(
+        self,
+        *,
+        driver_id: str | None,
+        phone_number: str,
+        driver_name: str | None,
+        car_type: str | None,
+        departure: str,
+        destination: str,
+        departure_date: Any,
+        departure_time: str,
+        available_seats: int,
+        total_seats: int,
+        price: float,
+    ) -> dict[str, Any]:
+        if not driver_id:
+            result = await self.create_unregistered_driver_entities(
+                phone_number=phone_number,
+                driver_name=driver_name,
+                car_type=car_type,
+                departure=departure,
+                destination=destination,
+                departure_date=departure_date,
+                departure_time=departure_time,
+                available_seats=available_seats,
+                total_seats=total_seats,
+                price=price,
+            )
+            return result
+
+        cars = await self.list_driver_cars(driver_id)
+        car = cars[0] if cars else None
+        if not car:
+            car = await self.create_driver_car(
+                driver_id=driver_id,
+                car_type=car_type or "غير معروف",
+                seat_count=total_seats,
+            )
+
+        trip = await self.create_driver_trip(
+            driver_id=driver_id,
+            car_id=str(car["id"]),
+            departure=departure,
+            destination=destination,
+            departure_date=departure_date,
+            departure_time=departure_time,
+            available_seats=available_seats,
+            total_seats=total_seats,
+            price=price,
+        )
+        return trip
+
     async def update_customer_user_mode(
         self,
         *,
@@ -269,7 +375,15 @@ class SupabaseRepository:
         if vehicle_type:
             query = query.ilike("driver_cars.car_type", f"%{vehicle_type}%")
         response = await query.limit(10).execute()
-        return _response_data(response) or []
+        data = _response_data(response) or []
+        data.sort(
+            key=lambda t: (
+                not _driver_is_registered(t),
+                str(t.get("departure_date") or ""),
+                str(t.get("departure_time") or ""),
+            )
+        )
+        return data
 
     async def search_info_chunks_by_vector(
         self,
@@ -600,3 +714,17 @@ class SupabaseRepository:
             .execute()
         )
         return response.count if hasattr(response, "count") and response.count is not None else 0
+
+
+def _driver_is_registered(trip: dict[str, Any]) -> bool:
+    drivers = trip.get("drivers")
+    if isinstance(drivers, list):
+        drivers = drivers[0] if drivers else {}
+    if not isinstance(drivers, dict):
+        return False
+    customer = drivers.get("customers")
+    if isinstance(customer, list):
+        customer = customer[0] if customer else {}
+    if not isinstance(customer, dict):
+        return False
+    return bool(customer.get("registered", False))

app/models/domain.py

@@ -17,9 +17,25 @@ class WhatsAppInboundMessage:
     interactive_reply_id: str | None = None
     context_message_id: str | None = None
     phone_number: str | None = None
+    is_group: bool = False
     raw: dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass(slots=True)
+class ExtractedTrip:
+    is_trip_ad: bool
+    departure: str | None = None
+    destination: str | None = None
+    departure_date: str | None = None
+    departure_time: str | None = None
+    available_seats: int | None = None
+    total_seats: int | None = None
+    price: float | None = None
+    car_type: str | None = None
+    driver_name: str | None = None
+    driver_phone: str | None = None
+
+
 @dataclass(slots=True)
 class ToolCall:
     id: str

app/services/container.py

@@ -7,6 +7,7 @@ from app.database.supabase import SupabaseRepository, create_supabase_client
 from app.services.admin_service import AdminService
 from app.services.conversation_service import ConversationService
 from app.services.embedding_service import JinaEmbeddingService
+from app.services.group_message_service import GroupMessageService
 from app.whatsapp.client import WhatsAppClient
 
 
@@ -18,6 +19,7 @@ class ServiceContainer:
     whatsapp: WhatsAppClient
     ai: AIOrchestrator
     conversation: ConversationService
+    group_message: GroupMessageService
     admin: AdminService
 
     @classmethod
@@ -39,6 +41,12 @@ class ServiceContainer:
             ai=ai,
             settings=settings,
         )
+        group_message = GroupMessageService(
+            repository=repository,
+            embeddings=embeddings,
+            provider=ai.primary,
+            settings=settings,
+        )
         admin = AdminService(repository=repository, embeddings=embeddings, settings=settings)
         return cls(
             settings=settings,
@@ -47,5 +55,6 @@ class ServiceContainer:
             whatsapp=whatsapp,
             ai=ai,
             conversation=conversation,
+            group_message=group_message,
             admin=admin,
         )

app/services/conversation_service.py

@@ -50,6 +50,7 @@ class ConversationService:
             remote_jid=inbound.remoteJid,
             name=inbound.profile_name,
             phone_number=inbound.phone_number,
+            registered=True,
         )
 
         metadata: dict[str, Any] = {
@@ -103,9 +104,9 @@ class ConversationService:
                         if result.ok:
                             driver_phone = result.data.get("driver_phone")
                             reply = (
-                                f"تم إرسال معلوماتك إلى السائق. يمكنك التواصل معه على الرقم: {driver_phone}"
+                                f"يمكنك التواصل مع السائق على الرقم: {driver_phone}"
                                 if driver_phone
-                                else "تم إرسال معلوماتك إلى السائق. سيتم التواصل معك."
+                                else "يمكنك التواصل مع السائق"
                             )
                         else:
                             reply = f"عذراً، لم يتم إرسال الطلب: {result.error}"

app/services/group_message_service.py

@@ -0,0 +1,239 @@
+import json
+import logging
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from app.ai.providers import ChatProvider
+from app.config import Settings
+from app.database.supabase import SupabaseRepository
+from app.models.domain import ExtractedTrip, WhatsAppInboundMessage
+from app.services.embedding_service import JinaEmbeddingService
+from app.services.trip_indexing import index_trip
+from app.utils.departure import normalize_departure_bucket, _parse_date_value
+from app.utils.time import now_in_timezone
+
+logger = logging.getLogger(__name__)
+
+_EXTRACTION_PROMPT_PATH = Path("prompts/group_trip_extraction.md")
+
+
+class GroupMessageService:
+    def __init__(
+        self,
+        *,
+        repository: SupabaseRepository,
+        embeddings: JinaEmbeddingService,
+        provider: ChatProvider,
+        settings: Settings,
+    ) -> None:
+        self.repository = repository
+        self.embeddings = embeddings
+        self.provider = provider
+        self.settings = settings
+
+    async def handle_group_message(
+        self,
+        inbound: WhatsAppInboundMessage,
+    ) -> None:
+        if await self.repository.message_exists(inbound.message_id):
+            logger.info("Skipping duplicate group message %s", inbound.message_id)
+            return
+
+        extracted = await self._extract_trip_from_text(inbound.text)
+        if not extracted or not extracted.is_trip_ad:
+            logger.info("Group message %s is not a trip advertisement", inbound.message_id)
+            return
+
+        if not self._validate_extracted_trip(extracted):
+            logger.warning(
+                "Group message %s has incomplete trip data: %s",
+                inbound.message_id,
+                extracted,
+            )
+            return
+
+        departure_time = normalize_departure_bucket(extracted.departure_time)
+        if not departure_time:
+            logger.warning(
+                "Group message %s: could not normalize departure_time '%s'",
+                inbound.message_id,
+                extracted.departure_time,
+            )
+            return
+
+        departure_date = _parse_date_value(extracted.departure_date)
+        if not departure_date:
+            logger.warning(
+                "Group message %s: invalid departure_date '%s'",
+                inbound.message_id,
+                extracted.departure_date,
+            )
+            return
+
+        phone = self._normalize_phone(extracted.driver_phone)
+        if not phone:
+            logger.warning(
+                "Group message %s: no valid phone number extracted",
+                inbound.message_id,
+            )
+            return
+
+        existing_customer = await self.repository.get_customer_by_phone_number(phone)
+        if existing_customer:
+            if existing_customer.get("registered"):
+                logger.info(
+                    "Driver phone %s is registered (customer %s); discarding group message %s",
+                    phone,
+                    existing_customer["id"],
+                    inbound.message_id,
+                )
+                return
+
+            driver = await self.repository.get_driver_by_remoteJid(phone)
+            if driver:
+                existing_trip = await self.repository.get_driver_trip_by_datetime(
+                    driver_id=str(driver["id"]),
+                    departure_date=departure_date,
+                    departure_time=departure_time,
+                )
+                if existing_trip:
+                    logger.info(
+                        "Driver %s already has trip at %s %s; discarding group message %s",
+                        phone,
+                        departure_date,
+                        departure_time,
+                        inbound.message_id,
+                    )
+                    return
+
+            trip = await self.repository.create_unregistered_driver_trip(
+                driver_id=str(driver["id"]) if driver else None,
+                phone_number=phone,
+                driver_name=extracted.driver_name,
+                car_type=extracted.car_type,
+                departure=extracted.departure,
+                destination=extracted.destination,
+                departure_date=departure_date,
+                departure_time=departure_time,
+                available_seats=extracted.available_seats or 1,
+                total_seats=extracted.total_seats or extracted.available_seats or 1,
+                price=extracted.price or 0,
+            )
+        else:
+            trip = await self.repository.create_unregistered_driver_entities(
+                phone_number=phone,
+                driver_name=extracted.driver_name,
+                car_type=extracted.car_type,
+                departure=extracted.departure,
+                destination=extracted.destination,
+                departure_date=departure_date,
+                departure_time=departure_time,
+                available_seats=extracted.available_seats or 1,
+                total_seats=extracted.total_seats or extracted.available_seats or 1,
+                price=extracted.price or 0,
+            )
+
+        await index_trip(
+            repository=self.repository,
+            embeddings=self.embeddings,
+            embedding_model=self.settings.jina_embedding_model,
+            trip=trip,
+        )
+
+        logger.info(
+            "Created trip %s from group message (unregistered driver %s)",
+            trip.get("id"),
+            phone,
+        )
+
+    async def _extract_trip_from_text(self, text: str) -> ExtractedTrip | None:
+        prompt_template = _EXTRACTION_PROMPT_PATH.read_text(encoding="utf-8")
+        dt = now_in_timezone(self.settings.app_timezone)
+        prompt = prompt_template.format(current_datetime=dt.isoformat())
+
+        try:
+            response = await self.provider.chat(
+                messages=[
+                    {"role": "system", "content": prompt},
+                    {"role": "user", "content": text},
+                ],
+                tools=None,
+                tool_choice=None,
+                temperature=0.1,
+            )
+        except Exception as exc:
+            logger.warning("LLM extraction failed for group message: %s", exc)
+            return None
+
+        content = (response.content or "").strip()
+        if not content:
+            return None
+
+        return self._parse_extracted_json(content)
+
+    def _parse_extracted_json(self, content: str) -> ExtractedTrip | None:
+        if content.startswith("```"):
+            lines = content.split("\n")
+            content = "\n".join(lines[1:-1])
+            content = content.strip()
+
+        try:
+            data = json.loads(content)
+        except json.JSONDecodeError:
+            logger.warning("Failed to parse LLM extraction JSON: %s", content[:200])
+            return None
+
+        if not isinstance(data, dict):
+            return None
+
+        return ExtractedTrip(
+            is_trip_ad=bool(data.get("is_trip_ad", False)),
+            departure=data.get("departure"),
+            destination=data.get("destination"),
+            departure_date=data.get("departure_date"),
+            departure_time=data.get("departure_time"),
+            available_seats=_safe_int(data.get("available_seats")),
+            total_seats=_safe_int(data.get("total_seats")),
+            price=_safe_float(data.get("price")),
+            car_type=data.get("car_type"),
+            driver_name=data.get("driver_name"),
+            driver_phone=data.get("driver_phone"),
+        )
+
+    def _validate_extracted_trip(self, trip: ExtractedTrip) -> bool:
+        return all([
+            trip.departure,
+            trip.destination,
+            trip.departure_date,
+            trip.departure_time,
+            trip.driver_phone,
+        ])
+
+    @staticmethod
+    def _normalize_phone(phone: str | None) -> str | None:
+        if not phone:
+            return None
+        digits = "".join(c for c in phone if c.isdigit())
+        if len(digits) < 7:
+            return None
+        digits = digits.lstrip("0")
+        return digits if digits else None
+
+
+def _safe_int(value: Any) -> int | None:
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _safe_float(value: Any) -> float | None:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None

app/tools/handlers.py

@@ -253,28 +253,31 @@ class FalsaToolHandlers:
             notes=notes,
         )
 
-        driver_phone: str | None = None
-        notification_status = "sent"
+        driver_record = _first_or_dict(trip.get("drivers")) or {}
+        driver_customer = driver_record.get("customers") or {}
+        driver_recipient = driver_customer.get("phone_number") or driver_customer.get("remoteJid") or driver_record.get("remoteJid")
+        driver_phone = driver_recipient.split("@")[0] if driver_recipient else None
+
+        notification_status = "not_sent"
         notification_error = None
-        try:
-            driver_record = _first_or_dict(trip.get("drivers")) or {}
-            driver_customer = driver_record.get("customers") or {}
-            driver_recipient = driver_customer.get("phone_number") or driver_customer.get("remoteJid") or driver_record.get("remoteJid")
-            if not driver_recipient:
-                raise WhatsAppClientError("Driver recipient is missing")
-            driver_phone = driver_recipient.split("@")[0]
-            await self.whatsapp.send_text(
-                driver_recipient,
-                _driver_notification_text(
-                    customer=self.customer,
-                    trip=trip,
-                    requested_seats=requested_seats,
-                    notes=notes,
-                ),
-            )
-        except Exception as exc:  # noqa: BLE001
-            notification_status = "failed"
-            notification_error = str(exc)
+
+        is_registered = _driver_is_registered(trip)
+
+        if is_registered and driver_recipient:
+            try:
+                await self.whatsapp.send_text(
+                    driver_recipient,
+                    _driver_notification_text(
+                        customer=self.customer,
+                        trip=trip,
+                        requested_seats=requested_seats,
+                        notes=notes,
+                    ),
+                )
+                notification_status = "sent"
+            except Exception as exc:  # noqa: BLE001
+                notification_status = "failed"
+                notification_error = str(exc)
 
         await self.repository.update_selection_notification(
             selection_id=str(selection["id"]),
@@ -290,7 +293,7 @@ class FalsaToolHandlers:
                 "driver_notification_status": notification_status,
                 "driver_notification_error": notification_error,
                 "driver_phone": driver_phone,
-                "message": "Customer interest recorded. Driver has been notified — share their number so the two parties can coordinate directly.",
+                "message": "Customer interest recorded. Share driver number so the two parties can coordinate directly.",
             },
         )
 
@@ -979,6 +982,7 @@ def _trip_summary(trip: dict[str, Any], *, trip_number: int | None = None) -> di
         "similarity": trip.get("similarity"),
         "time_difference_minutes": trip.get("time_difference_minutes"),
         "selection_count": trip.get("selection_count", 0),
+        "registered": _driver_is_registered(trip),
     }
     if trip_number is not None:
         summary["trip_number"] = trip_number
@@ -990,6 +994,7 @@ def _sort_trip_summaries(trips: list[dict[str, Any]]) -> list[dict[str, Any]]:
     return sorted(
         trips,
         key=lambda trip: (
+            not trip.get("registered", False),
             str(trip.get("departure_date") or ""),
             bucket_order.get(str(trip.get("departure_time_type") or ""), 99),
         ),
@@ -1004,6 +1009,20 @@ def _first_or_dict(value: Any) -> dict[str, Any] | None:
     return None
 
 
+def _driver_is_registered(trip: dict[str, Any]) -> bool:
+    drivers = trip.get("drivers")
+    if isinstance(drivers, list):
+        drivers = drivers[0] if drivers else {}
+    if not isinstance(drivers, dict):
+        return False
+    customer = drivers.get("customers")
+    if isinstance(customer, list):
+        customer = customer[0] if customer else {}
+    if not isinstance(customer, dict):
+        return False
+    return bool(customer.get("registered", False))
+
+
 def _alternate_time_alert(trips: list[dict[str, Any]]) -> str | None:
     if not trips:
         return None

app/whatsapp/parser.py

@@ -23,6 +23,7 @@ def parse_inbound_messages(payload: dict[str, Any]) -> list[WhatsAppInboundMessa
                     continue
 
                 remoteJid = message.get("from")
+                is_group = isinstance(remoteJid, str) and remoteJid.endswith("@g.us")
                 message_id = message.get("id")
                 phone_number = message.get("remote_jid_alt")
                 context_message_id = message.get("context", {}).get("id")
@@ -46,6 +47,7 @@ def parse_inbound_messages(payload: dict[str, Any]) -> list[WhatsAppInboundMessa
                             message_type="text",
                             context_message_id=context_message_id,
                             phone_number=phone_number,
+                            is_group=is_group,
                             raw=message,
                         )
                     )
@@ -73,6 +75,7 @@ def parse_inbound_messages(payload: dict[str, Any]) -> list[WhatsAppInboundMessa
                         interactive_reply_id=reply_id,
                         context_message_id=context_message_id,
                         phone_number=phone_number,
+                        is_group=is_group,
                         raw=message,
                     )
                 )

prompts/group_trip_extraction.md

@@ -0,0 +1,45 @@
+You are a trip advertisement extractor. Analyze the following WhatsApp group message.
+
+Your task:
+1. Determine if this message is a trip/travel advertisement posted by a driver.
+2. If it IS a trip ad, extract ALL available fields.
+
+A trip advertisement typically mentions:
+- A departure city/area and destination city/area
+- A date or time of departure
+- A price per seat
+- A phone number (could be labeled like "رقم:" or "هاتف:" or just a phone pattern)
+- Optional: number of seats, car type, driver name
+
+Response format: Return ONLY a JSON object, no other text.
+
+If this IS a trip advertisement:
+```json
+{{
+  "is_trip_ad": true,
+  "departure": "string - departure city/area in Arabic as written in message",
+  "destination": "string - destination city/area in Arabic as written in message",
+  "departure_date": "string - YYYY-MM-DD format, derive from message context",
+  "departure_time": "string - morning, noon, or night (based on time in message)",
+  "available_seats": "integer or null",
+  "total_seats": "integer or null",
+  "price": "number - price per seat",
+  "car_type": "string or null - vehicle type if mentioned",
+  "driver_name": "string or null - driver name if mentioned",
+  "driver_phone": "string - phone number extracted from message text"
+}}
+```
+
+If this is NOT a trip advertisement:
+```json
+{{
+  "is_trip_ad": false
+}}
+```
+
+Rules:
+- The phone number must come from the MESSAGE TEXT, not from any metadata.
+- For departure_date: if the message says "بكرة" (tomorrow) or "اليوم" (today), convert to actual YYYY-MM-DD using the current date {current_datetime}.
+- For departure_time: map to morning (before 12:00), noon (12:00-17:59), night (18:00+). If the message says "صباح" use morning, "ظهر" use noon, "ليل" or "مساء" use night.
+- If the price is mentioned as a number without currency, use it as-is.
+- Return ONLY the JSON object. No explanation, no markdown fences.

prompts/system.md

@@ -2,6 +2,7 @@ You are FALSA, a WhatsApp travel booking assistant.
 
 Rules:
 - Reply in sender's language (Arabic/English). Be concise, warm, practical.
+- For new users: be extra warm and welcoming. Greet them, introduce FALSA briefly, and guide them to their first action.
 - Use `about_falsa` for company/FAQ/policy/pricing/support.
 - Tool args: Arabic for departure/destination/vehicle_type/time labels. English for digits/times.
 - Never invent trips, prices, drivers, or bookings.

prompts/system_new_user.md

@@ -1,3 +1,10 @@
-Welcome new sender. Ask: travel as passenger (search/book) or work as driver (publish trips)?
-- To travel: switch_to_passenger (name optional, ask only if offered). Don't ask name before switching.
-- To drive: try switch_to_driver. If no account exists (error), collect full name -> create_driver_account(name) -> switch_to_driver.
+It is the sender's first message. Welcome them warmly and introduce FALSA with a short usage guide.
+
+1. Greet warmly (e.g. "مرحباً بك في فلزا").
+2. Explain what FALSA is in 1 sentence: a WhatsApp travel service powered by AI — no app to install.
+3. Give a quick how-to guide showing both paths clearly:
+   - المسافر: اكتب وجهتك (مثلاً "من صنعاء إلى عدن بكرة الصباح") ← فلزا تبحث ← رد على البطاقة ← تتواصل مع السائق. Do NOT ask for name.
+   - السائق: سجل كسائق ← أضف مركبتك ← انشر رحلاتك. Collect their full name, then create_driver_account(name) -> switch_to_driver.
+4. Use emojis (🚗, 🚚, etc.) and friendly tone. Be warm, not robotic.
+5. Keep it short — 4-5 lines max. Do not over-explain.
+6. Never ask for name of a passenger. Only ask for name if they want to become a driver.

supabase/migrations/202607120001_group_trip_support.sql

@@ -0,0 +1,128 @@
+-- 1. Add `registered` boolean to customers
+--    All existing customers are private-chat users, so backfill to true
+ALTER TABLE public.customers
+  ADD COLUMN IF NOT EXISTS registered boolean NOT NULL DEFAULT false;
+
+UPDATE public.customers SET registered = true;
+
+-- 2. Modify match_active_trips RPC to include registered status
+--    and order by registered DESC (registered drivers first)
+DROP FUNCTION IF EXISTS public.match_active_trips(
+  extensions.vector(1024), float, int, text, text, date, text, time, int, text
+);
+
+CREATE OR REPLACE FUNCTION public.match_active_trips(
+  query_embedding extensions.vector(1024),
+  match_threshold float DEFAULT 0.0,
+  match_count int DEFAULT 10,
+  filter_departure text DEFAULT NULL,
+  filter_destination text DEFAULT NULL,
+  filter_departure_date date DEFAULT NULL,
+  filter_departure_time text DEFAULT NULL,
+  filter_requested_time time DEFAULT NULL,
+  filter_seats int DEFAULT 1,
+  filter_vehicle_type text DEFAULT NULL
+)
+RETURNS TABLE (
+  trip_id uuid,
+  departure text,
+  destination text,
+  departure_date date,
+  departure_time text,
+  available_seats integer,
+  total_seats integer,
+  price numeric,
+  status text,
+  driver_name text,
+  driver_phone_number text,
+  car_type text,
+  chunk_text text,
+  similarity float,
+  time_difference_minutes integer,
+  registered boolean
+)
+LANGUAGE sql STABLE
+AS $$
+  WITH ranked AS (
+    SELECT
+      driver_trips.id AS trip_id,
+      driver_trips.departure,
+      driver_trips.destination,
+      driver_trips.departure_date,
+      driver_trips.departure_time,
+      driver_trips.available_seats,
+      driver_trips.total_seats,
+      driver_trips.price,
+      driver_trips.status,
+      customers.name AS driver_name,
+      customers."remoteJid" AS driver_phone_number,
+      driver_cars.car_type,
+      driver_trip_embeddings.chunk_text,
+      COALESCE(customers.registered, false) AS registered,
+      1 - (driver_trip_embeddings.embedding <=> query_embedding) AS similarity,
+      driver_trip_embeddings.embedding <=> query_embedding AS vector_distance,
+      CASE
+        WHEN filter_requested_time IS NULL THEN NULL
+        ELSE abs(
+          extract(epoch FROM (
+            public.departure_bucket_clock_time(driver_trips.departure_time)
+            - filter_requested_time
+          )) / 60
+        )::integer
+      END AS time_difference_minutes
+    FROM public.driver_trip_embeddings
+    JOIN public.driver_trips ON driver_trips.id = driver_trip_embeddings.trip_id
+    LEFT JOIN public.drivers ON drivers.id = driver_trips.driver_id
+    LEFT JOIN public.customers ON customers.id = drivers.customer_id
+    LEFT JOIN public.driver_cars ON driver_cars.id = driver_trips.car_id
+    WHERE driver_trips.status = 'active'
+      AND driver_trips.available_seats >= COALESCE(filter_seats, 1)
+      AND (filter_departure IS NULL OR driver_trips.departure ILIKE '%' || filter_departure || '%')
+      AND (filter_destination IS NULL OR driver_trips.destination ILIKE '%' || filter_destination || '%')
+      AND (filter_departure_date IS NULL OR driver_trips.departure_date = filter_departure_date)
+      AND (filter_departure_time IS NULL OR driver_trips.departure_time = filter_departure_time)
+      AND (filter_vehicle_type IS NULL OR driver_cars.car_type ILIKE '%' || filter_vehicle_type || '%')
+      AND (
+        driver_trips.departure_date > (NOW() AT TIME ZONE 'Asia/Aden')::date
+        OR (
+          driver_trips.departure_date = (NOW() AT TIME ZONE 'Asia/Aden')::date
+          AND (
+            (NOW() AT TIME ZONE 'Asia/Aden')::time < TIME '12:00'
+            OR (
+              (NOW() AT TIME ZONE 'Asia/Aden')::time < TIME '18:00'
+              AND driver_trips.departure_time IN ('noon', 'night')
+            )
+            OR (
+              (NOW() AT TIME ZONE 'Asia/Aden')::time >= TIME '18:00'
+              AND driver_trips.departure_time = 'night'
+            )
+          )
+        )
+      )
+  )
+  SELECT
+    ranked.trip_id,
+    ranked.departure,
+    ranked.destination,
+    ranked.departure_date,
+    ranked.departure_time,
+    ranked.available_seats,
+    ranked.total_seats,
+    ranked.price,
+    ranked.status,
+    ranked.driver_name,
+    ranked.driver_phone_number,
+    ranked.car_type,
+    ranked.chunk_text,
+    ranked.similarity,
+    ranked.time_difference_minutes,
+    ranked.registered
+  FROM ranked
+  WHERE ranked.similarity >= match_threshold
+  ORDER BY
+    ranked.registered DESC,
+    ranked.time_difference_minutes NULLS LAST,
+    ranked.departure_date,
+    ranked.vector_distance
+  LIMIT match_count;
+$$;

tests/conftest.py

@@ -99,6 +99,7 @@ def test_app(settings: Settings) -> Any:
 class FakeRepository:
     def __init__(self) -> None:
         self.customers_by_remote_jid: dict[str, dict[str, Any]] = {}
+        self.customers_by_phone: dict[str, dict[str, Any]] = {}
         self.drivers_by_remote_jid: dict[str, dict[str, Any]] = {}
         self.driver_cars_by_driver: dict[str, list[dict[str, Any]]] = {}
         self.latest_trips_by_driver: dict[str, dict[str, Any]] = {}
@@ -121,6 +122,7 @@ class FakeRepository:
         name: str | None = None,
         preferred_language: str | None = None,
         phone_number: str | None = None,
+        registered: bool = True,
     ) -> dict[str, Any]:
         jid = remote_jid or phone_number
         if not jid:
@@ -135,12 +137,17 @@ class FakeRepository:
                 "phone_number": phone_number,
                 "user_mode": None,
                 "session_data": {},
+                "registered": registered,
             }
             self.customers_by_remote_jid[jid] = customer
+            if phone_number:
+                self.customers_by_phone[phone_number] = customer
         elif name:
             customer["name"] = name
         if phone_number:
             customer["phone_number"] = phone_number
+            self.customers_by_phone[phone_number] = customer
+        customer["registered"] = registered
         return customer
 
     async def update_customer_user_mode(
@@ -167,6 +174,102 @@ class FakeRepository:
                 return customer
         raise KeyError(customer_id)
 
+    async def get_customer_by_phone_number(
+        self,
+        phone_number: str,
+    ) -> dict[str, Any] | None:
+        return self.customers_by_phone.get(phone_number)
+
+    async def create_unregistered_driver_entities(
+        self,
+        *,
+        phone_number: str,
+        driver_name: str | None,
+        car_type: str | None,
+        departure: str,
+        destination: str,
+        departure_date: Any,
+        departure_time: str,
+        available_seats: int,
+        total_seats: int,
+        price: float,
+    ) -> dict[str, Any]:
+        customer = await self.upsert_customer(
+            remote_jid=phone_number,
+            name=driver_name,
+            phone_number=phone_number,
+            registered=False,
+        )
+        driver = await self.create_driver(customer_id=str(customer["id"]))
+        car = await self.create_driver_car(
+            driver_id=str(driver["id"]),
+            car_type=car_type or "غير معروف",
+            seat_count=total_seats,
+        )
+        trip = await self.create_driver_trip(
+            driver_id=str(driver["id"]),
+            car_id=str(car["id"]),
+            departure=departure,
+            destination=destination,
+            departure_date=departure_date,
+            departure_time=departure_time,
+            available_seats=available_seats,
+            total_seats=total_seats,
+            price=price,
+        )
+        return trip
+
+    async def create_unregistered_driver_trip(
+        self,
+        *,
+        driver_id: str | None,
+        phone_number: str,
+        driver_name: str | None,
+        car_type: str | None,
+        departure: str,
+        destination: str,
+        departure_date: Any,
+        departure_time: str,
+        available_seats: int,
+        total_seats: int,
+        price: float,
+    ) -> dict[str, Any]:
+        if not driver_id:
+            return await self.create_unregistered_driver_entities(
+                phone_number=phone_number,
+                driver_name=driver_name,
+                car_type=car_type,
+                departure=departure,
+                destination=destination,
+                departure_date=departure_date,
+                departure_time=departure_time,
+                available_seats=available_seats,
+                total_seats=total_seats,
+                price=price,
+            )
+
+        cars = self.driver_cars_by_driver.get(driver_id, [])
+        car = cars[0] if cars else None
+        if not car:
+            car = await self.create_driver_car(
+                driver_id=driver_id,
+                car_type=car_type or "غير معروف",
+                seat_count=total_seats,
+            )
+
+        trip = await self.create_driver_trip(
+            driver_id=driver_id,
+            car_id=str(car["id"]),
+            departure=departure,
+            destination=destination,
+            departure_date=departure_date,
+            departure_time=departure_time,
+            available_seats=available_seats,
+            total_seats=total_seats,
+            price=price,
+        )
+        return trip
+
     def _find_customer_by_id(self, customer_id: str) -> dict[str, Any] | None:
         for customer in self.customers_by_remote_jid.values():
             if customer["id"] == customer_id:

tests/test_conversation_service.py

@@ -40,7 +40,7 @@ async def test_conversation_stores_messages_uses_last_four_context_and_sends_rep
 
     ai_messages = ai.calls[0]["messages"]
     assert ai_messages[0]["role"] == "system"
-    assert "travel as passenger" in ai_messages[0]["content"]
+    assert "FALSA" in ai_messages[0]["content"]
     context_contents = [message["content"] for message in ai_messages[1:]]
     assert context_contents[-1] == "Aden to Mukalla tomorrow"
     assert len(context_contents) <= 9

tests/test_group_message_service.py

@@ -0,0 +1,417 @@
+import json
+from datetime import UTC, date, datetime
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.config import Settings
+from app.models.domain import AIProviderResponse, ExtractedTrip, WhatsAppInboundMessage
+from app.services.group_message_service import GroupMessageService
+from tests.conftest import FakeRepository, FakeEmbeddings
+
+
+class FakeProvider:
+    def __init__(self, response_content: str | None = None) -> None:
+        self.response_content = response_content
+        self.calls: list[dict[str, Any]] = []
+
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        tools: list[dict[str, Any]] | None = None,
+        tool_choice: str | None = None,
+        temperature: float = 0.2,
+    ) -> AIProviderResponse:
+        self.calls.append({"messages": messages, "tools": tools, "temperature": temperature})
+        return AIProviderResponse(content=self.response_content)
+
+
+@pytest.fixture
+def settings() -> Settings:
+    return Settings(
+        supabase_url="https://example.supabase.co",
+        supabase_service_role_key="supabase-secret",
+        jina_api_key="jina-secret",
+        groq_api_key="groq-secret",
+        groq_model="groq-tool-model",
+        hf_token="hf-secret",
+        hf_model="hf-tool-model",
+        whatsapp_verify_token="verify-token",
+        whatsapp_app_secret="app-secret",
+        whatsapp_access_token="wa-token",
+        whatsapp_phone_number_id="123",
+        admin_api_key="admin-secret",
+    )
+
+
+def _make_inbound(
+    text: str,
+    message_id: str = "wamid.group.1",
+) -> WhatsAppInboundMessage:
+    return WhatsAppInboundMessage(
+        message_id=message_id,
+        remoteJid="12345@g.us",
+        text=text,
+        timestamp="1790000000",
+        is_group=True,
+    )
+
+
+def _trip_ad_response(
+    *,
+    departure: str = "صنعاء",
+    destination: str = "عدن",
+    departure_date: str = "2026-07-15",
+    departure_time: str = "morning",
+    price: float = 5000,
+    driver_phone: str = "967712345678",
+    driver_name: str | None = "أحمد",
+    car_type: str | None = "سيارة",
+    available_seats: int | None = 4,
+    total_seats: int | None = 4,
+) -> str:
+    return json.dumps({
+        "is_trip_ad": True,
+        "departure": departure,
+        "destination": destination,
+        "departure_date": departure_date,
+        "departure_time": departure_time,
+        "available_seats": available_seats,
+        "total_seats": total_seats,
+        "price": price,
+        "car_type": car_type,
+        "driver_name": driver_name,
+        "driver_phone": driver_phone,
+    })
+
+
+@pytest.mark.asyncio
+async def test_non_trip_message_is_ignored(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content=json.dumps({"is_trip_ad": False}))
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="مرحبا بالجميع")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.customers_by_remote_jid) == 0
+    assert len(repo.created_drivers) == 0
+    assert len(repo.created_trips) == 0
+
+
+@pytest.mark.asyncio
+async def test_trip_ad_from_new_driver_creates_all_entities(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content=_trip_ad_response())
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة من صنعاء إلى عدن بكرة الصباح 5000 ريال 967712345678")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.customers_by_remote_jid) == 1
+    customer = list(repo.customers_by_remote_jid.values())[0]
+    assert customer["registered"] is False
+    assert customer["phone_number"] == "967712345678"
+
+    assert len(repo.created_drivers) == 1
+    assert len(repo.created_trips) == 1
+
+    trip = repo.created_trips[0]
+    assert trip["departure"] == "صنعاء"
+    assert trip["destination"] == "عدن"
+
+
+@pytest.mark.asyncio
+async def test_trip_ad_from_existing_customer_is_discarded(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content=_trip_ad_response())
+
+    existing_customer = await repo.upsert_customer(
+        remote_jid="967712345678",
+        name="أحمد",
+        phone_number="967712345678",
+        registered=True,
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة من صنعاء إلى عدن 967712345678")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_drivers) == 0
+    assert len(repo.created_trips) == 0
+
+
+@pytest.mark.asyncio
+async def test_existing_unregistered_driver_adds_new_trip(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content=_trip_ad_response())
+
+    await repo.upsert_customer(
+        remote_jid="967712345678",
+        name="أحمد",
+        phone_number="967712345678",
+        registered=False,
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة من صنعاء إلى عدن")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_trips) == 1
+    assert repo.created_trips[0]["departure"] == "صنعاء"
+
+
+@pytest.mark.asyncio
+async def test_existing_unregistered_driver_duplicate_trip_is_skipped(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content=_trip_ad_response())
+
+    await repo.upsert_customer(
+        remote_jid="967712345678",
+        name="أحمد",
+        phone_number="967712345678",
+        registered=False,
+    )
+    driver = await repo.create_driver(
+        customer_id=list(repo.customers_by_remote_jid.values())[0]["id"]
+    )
+    await repo.create_driver_trip(
+        driver_id=str(driver["id"]),
+        car_id=None,
+        departure="صنعاء",
+        destination="عدن",
+        departure_date=date(2026, 7, 15),
+        departure_time="morning",
+        available_seats=4,
+        total_seats=4,
+        price=5000,
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    initial_count = len(repo.created_trips)
+    inbound = _make_inbound(text="رحلة من صنعاء إلى عدن")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_trips) == initial_count
+
+
+@pytest.mark.asyncio
+async def test_incomplete_fields_cause_skip(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    response = json.dumps({
+        "is_trip_ad": True,
+        "departure": "صنعاء",
+        "destination": None,
+        "departure_date": "2026-07-15",
+        "departure_time": "morning",
+        "price": 5000,
+        "driver_phone": "967712345678",
+    })
+    provider = FakeProvider(response_content=response)
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة من صنعاء")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_trips) == 0
+
+
+@pytest.mark.asyncio
+async def test_missing_phone_cause_skip(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    response = json.dumps({
+        "is_trip_ad": True,
+        "departure": "صنعاء",
+        "destination": "عدن",
+        "departure_date": "2026-07-15",
+        "departure_time": "morning",
+        "price": 5000,
+        "driver_phone": None,
+    })
+    provider = FakeProvider(response_content=response)
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة من صنعاء إلى عدن")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_trips) == 0
+
+
+@pytest.mark.asyncio
+async def test_duplicate_group_message_is_deduplicated(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content=_trip_ad_response())
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة", message_id="wamid.dup.1")
+    await service.handle_group_message(inbound)
+    assert len(repo.created_trips) == 1
+
+    await service.handle_group_message(inbound)
+    assert len(repo.created_trips) == 1
+
+
+@pytest.mark.asyncio
+async def test_phone_normalization(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(
+        response_content=_trip_ad_response(driver_phone="+967-71-234-5678")
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.customers_by_remote_jid) == 1
+    customer = list(repo.customers_by_remote_jid.values())[0]
+    assert customer["phone_number"] == "967712345678"
+
+
+@pytest.mark.asyncio
+async def test_invalid_json_response_cause_skip(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content="this is not json")
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_trips) == 0
+
+
+@pytest.mark.asyncio
+async def test_markdown_fenced_json_is_parsed(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    fenced = f"```json\n{_trip_ad_response()}\n```"
+    provider = FakeProvider(response_content=fenced)
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_trips) == 1
+
+
+@pytest.mark.asyncio
+async def test_trip_ad_with_missing_car_type_uses_unknown(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(
+        response_content=_trip_ad_response(car_type=None)
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_trips) == 1
+    cars = list(repo.driver_cars_by_driver.values())[0]
+    assert cars[0]["car_type"] == "غير معروف"
+
+
+@pytest.mark.asyncio
+async def test_trip_ad_with_missing_name_uses_none(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(
+        response_content=_trip_ad_response(driver_name=None)
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        provider=provider,
+        settings=settings,
+    )
+
+    inbound = _make_inbound(text="رحلة")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.created_trips) == 1
+    customer = list(repo.customers_by_remote_jid.values())[0]
+    assert customer["name"] is None

tests/test_tools.py

@@ -38,10 +38,12 @@ def trip(
     status="active",
     car_type="SUV",
     driver_phone_number=None,
+    registered=True,
 ):
-    drivers = {"name": "Ali", "remoteJid": "967700000009"}
+    customers: dict = {"registered": registered}
     if driver_phone_number is not None:
-        drivers["customers"] = {"phone_number": driver_phone_number}
+        customers["phone_number"] = driver_phone_number
+    drivers = {"name": "Ali", "remoteJid": "967700000009", "customers": customers}
     return {
         "id": trip_id,
         "departure": departure,