feat: support multiple phone numbers and prevent duplicate customers

- Add support for multiple phone numbers separated by / in driver profiles
- Normalize phone numbers with country code prepending and leading zero stripping
- Prevent duplicate customers by checking phone_number before upserting
- Merge customer records when same phone appears in group vs DM context
- Make remote_jid nullable for unregistered drivers discovered via groups
- Add database migrations for nullable remote_jid and updated RPC

app/config.py

@@ -47,6 +47,8 @@ class Settings(BaseSettings):
 
     admin_api_key: str = Field(min_length=1)
 
+    default_country_code: str = "967"
+
 
 @lru_cache
 def get_settings() -> Settings:

app/database/supabase.py

@@ -35,7 +35,7 @@ class SupabaseRepository:
     async def upsert_customer(
         self,
         *,
-        remote_jid: str,
+        remote_jid: str | None = None,
         name: str | None = None,
         preferred_language: str | None = None,
         phone_number: str | None = None,
@@ -43,6 +43,33 @@ class SupabaseRepository:
     ) -> dict[str, Any]:
         if phone_number:
             phone_number = phone_number.split("@")[0]
+
+        # Always check by phone_number first to avoid duplicates.
+        # A driver may have been saved from a group (remoteJid=None, phone_number=X),
+        # then later contacts us privately (remoteJid=Y, phone_number=X).
+        if phone_number:
+            existing = await self.get_customer_by_phone_number(phone_number)
+            if existing:
+                update_payload: dict[str, Any] = {}
+                if remote_jid and not existing.get("remoteJid"):
+                    update_payload["remoteJid"] = remote_jid
+                if name is not None:
+                    update_payload["name"] = name
+                if registered is not None:
+                    update_payload["registered"] = registered
+                if phone_number is not None:
+                    update_payload["phone_number"] = phone_number
+                if update_payload:
+                    response = await (
+                        self.client.table("customers")
+                        .update(update_payload)
+                        .eq("id", existing["id"])
+                        .execute()
+                    )
+                    data = _response_data(response)
+                    return data[0] if isinstance(data, list) else data
+                return existing
+
         payload = {
             "remoteJid": remote_jid,
             "name": name,
@@ -63,6 +90,7 @@ class SupabaseRepository:
         self,
         phone_number: str,
     ) -> dict[str, Any] | None:
+        # Try exact match first
         response = await (
             self.client.table("customers")
             .select("*")
@@ -70,7 +98,38 @@ class SupabaseRepository:
             .maybe_single()
             .execute()
         )
-        return _response_data(response)
+        result = _response_data(response)
+        if result:
+            return result
+
+        # Try each individual phone from /-separated incoming number
+        phones = [p.strip() for p in phone_number.split("/")] if "/" in phone_number else [phone_number]
+        for phone in phones:
+            response = await (
+                self.client.table("customers")
+                .select("*")
+                .eq("phone_number", phone)
+                .maybe_single()
+                .execute()
+            )
+            result = _response_data(response)
+            if result:
+                return result
+
+        # Check if any stored customer has a /-separated phone_number containing our phone
+        for phone in phones:
+            response = await (
+                self.client.table("customers")
+                .select("*")
+                .like("phone_number", f"%{phone}%")
+                .maybe_single()
+                .execute()
+            )
+            result = _response_data(response)
+            if result:
+                return result
+
+        return None
 
     async def create_unregistered_driver_entities(
         self,
@@ -87,7 +146,7 @@ class SupabaseRepository:
         price: float,
     ) -> dict[str, Any]:
         customer = await self.upsert_customer(
-            remote_jid=phone_number,
+            remote_jid=None,
             name=driver_name,
             phone_number=phone_number,
             registered=False,
@@ -544,6 +603,36 @@ class SupabaseRepository:
         )
         return _response_data(response)
 
+    async def get_driver_by_phone_number(
+        self,
+        phone_number: str,
+    ) -> dict[str, Any] | None:
+        """Look up a driver by phone_number column.
+
+        Supports multiple phones separated by '/': tries each one until a match is found.
+        """
+        if "/" in phone_number:
+            phones = phone_number.split("/")
+            for phone in phones:
+                result = await self._get_driver_by_single_phone(phone.strip())
+                if result:
+                    return result
+            return None
+        return await self._get_driver_by_single_phone(phone_number)
+
+    async def _get_driver_by_single_phone(
+        self,
+        phone: str,
+    ) -> dict[str, Any] | None:
+        response = await (
+            self.client.table("drivers")
+            .select("*, customers!inner(*)")
+            .eq("customers.phone_number", phone)
+            .maybe_single()
+            .execute()
+        )
+        return _response_data(response)
+
     async def create_driver(self, *, customer_id: str) -> dict[str, Any]:
         response = await (
             self.client.table("drivers")

app/services/group_message_service.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import re
 from datetime import date
 from pathlib import Path
 from typing import Any
@@ -71,7 +72,10 @@ class GroupMessageService:
             )
             return
 
-        phone = self._normalize_phone(extracted.driver_phone)
+        phone = self._normalize_phone(
+            extracted.driver_phone,
+            country_code=self.settings.default_country_code,
+        )
         if not phone:
             logger.warning(
                 "Group message %s: no valid phone number extracted",
@@ -90,7 +94,7 @@ class GroupMessageService:
                 )
                 return
 
-            driver = await self.repository.get_driver_by_remoteJid(phone)
+            driver = await self.repository.get_driver_by_phone_number(phone)
             if driver:
                 existing_trip = await self.repository.get_driver_trip_by_datetime(
                     driver_id=str(driver["id"]),
@@ -211,14 +215,29 @@ class GroupMessageService:
         ])
 
     @staticmethod
-    def _normalize_phone(phone: str | None) -> str | None:
+    def _normalize_phone(phone: str | None, country_code: str = "967") -> str | None:
         if not phone:
             return None
-        digits = "".join(c for c in phone if c.isdigit())
-        if len(digits) < 7:
+
+        # Handle multiple phone numbers separated by /, or ,
+        phones = re.split(r'[/,]', phone)
+        normalized_phones: list[str] = []
+
+        for p in phones:
+            digits = "".join(c for c in p if c.isdigit())
+            if len(digits) >= 7:
+                digits = digits.lstrip("0")
+                if digits:
+                    # Prepend country code if number doesn't already have it
+                    if not digits.startswith(country_code):
+                        digits = country_code + digits
+                    normalized_phones.append(digits)
+
+        if not normalized_phones:
             return None
-        digits = digits.lstrip("0")
-        return digits if digits else None
+
+        # Return single phone or multiple phones separated by /
+        return "/".join(normalized_phones) if len(normalized_phones) > 1 else normalized_phones[0]
 
 
 def _safe_int(value: Any) -> int | None:

app/tools/handlers.py

@@ -264,8 +264,17 @@ class FalsaToolHandlers:
 
         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
+
+        # For unregistered drivers, phone_number may contain multiple numbers separated by /
+        driver_phone_raw = driver_customer.get("phone_number")
+        if driver_phone_raw and "/" in driver_phone_raw:
+            # Multiple phone numbers - return them separated by /
+            driver_phone = driver_phone_raw
+            driver_recipient = driver_phone_raw.split("/")[0]
+        else:
+            # Single phone number - use existing logic
+            driver_recipient = driver_phone_raw 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

prompts/group_trip_extraction.md

@@ -26,7 +26,7 @@ If this IS a trip advertisement:
   "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"
+  "driver_phone": "string - ALL phone numbers found in the message, separated by / if multiple. Extract from message text only."
 }}
 ```
 

scripts/run_migration.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+"""Run the driver_message_card migration against Supabase.
+
+Usage:
+    source .venv/bin/activate
+    python scripts/run_migration.py
+
+Requires DATABASE_URL in .env or exports it.
+The DATABASE_URL should be the direct connection string from:
+    Supabase Dashboard → Settings → Database → Connection string → URI (Session mode)
+"""
+
+import os
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+try:
+    import psycopg
+except ImportError:
+    sys.exit("psycopg not installed. Run: pip install 'psycopg[binary]'")
+
+MIGRATION_SQL = (Path("supabase/migrations/202607140001_driver_message_card.sql")).read_text()
+
+
+def main():
+    db_url = os.environ.get("DATABASE_URL")
+    if not db_url:
+        sys.exit(
+            "DATABASE_URL not set. Add it to .env or export it.\n"
+            "Get it from: Supabase Dashboard → Settings → Database → Connection string → URI"
+        )
+
+    print(f"Connecting to database...")
+    conn = psycopg.connect(db_url, connect_timeout=10)
+    try:
+        with conn.cursor() as cur:
+            cur.execute(MIGRATION_SQL)
+            conn.commit()
+        print("Migration applied successfully!")
+    except Exception as e:
+        conn.rollback()
+        sys.exit(f"Migration failed: {e}")
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()

supabase/migrations/202607160001_make_remote_jid_nullable.sql

@@ -0,0 +1,6 @@
+-- Make remoteJid nullable so unregistered drivers from groups
+-- can be created without a WhatsApp JID (they only have phone_number)
+ALTER TABLE public.customers ALTER COLUMN "remoteJid" DROP NOT NULL;
+
+-- Update unique constraint to allow NULL values (PostgreSQL allows multiple NULLs by default)
+-- No additional action needed since UNIQUE already permits multiple NULLs

supabase/migrations/202607160002_update_rpc_phone_number.sql

@@ -0,0 +1,123 @@
+-- Update match_active_trips RPC to use phone_number instead of remoteJid
+-- This ensures unregistered drivers from groups (who have phone_number but no remoteJid)
+-- are properly exposed in trip search results
+
+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,
+      COALESCE(customers.phone_number, 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

@@ -124,6 +124,20 @@ class FakeRepository:
         phone_number: str | None = None,
         registered: bool = True,
     ) -> dict[str, Any]:
+        # Always check by phone_number first to avoid duplicates
+        if phone_number:
+            existing = await self.get_customer_by_phone_number(phone_number)
+            if existing:
+                if remote_jid and not existing.get("remoteJid"):
+                    existing["remoteJid"] = remote_jid
+                if name is not None:
+                    existing["name"] = name
+                if registered is not None:
+                    existing["registered"] = registered
+                if phone_number is not None:
+                    existing["phone_number"] = phone_number
+                return existing
+
         jid = remote_jid or phone_number
         if not jid:
             raise ValueError("remote_jid or phone_number is required")
@@ -131,7 +145,7 @@ class FakeRepository:
         if customer is None:
             customer = {
                 "id": f"cust-{len(self.customers_by_remote_jid) + 1}",
-                "remoteJid": jid,
+                "remoteJid": remote_jid,
                 "name": name,
                 "preferred_language": preferred_language,
                 "phone_number": phone_number,
@@ -178,7 +192,26 @@ class FakeRepository:
         self,
         phone_number: str,
     ) -> dict[str, Any] | None:
-        return self.customers_by_phone.get(phone_number)
+        # Try exact match first
+        result = self.customers_by_phone.get(phone_number)
+        if result:
+            return result
+
+        # Try each individual phone from /-separated incoming number
+        phones = [p.strip() for p in phone_number.split("/")] if "/" in phone_number else [phone_number]
+        for phone in phones:
+            result = self.customers_by_phone.get(phone)
+            if result:
+                return result
+
+        # Check if any stored customer has a /-separated phone_number containing our phone
+        for customer in self.customers_by_remote_jid.values():
+            stored = customer.get("phone_number") or ""
+            for phone in phones:
+                if phone in stored.split("/"):
+                    return customer
+
+        return None
 
     async def create_unregistered_driver_entities(
         self,
@@ -195,7 +228,7 @@ class FakeRepository:
         price: float,
     ) -> dict[str, Any]:
         customer = await self.upsert_customer(
-            remote_jid=phone_number,
+            remote_jid=None,
             name=driver_name,
             phone_number=phone_number,
             registered=False,
@@ -424,6 +457,36 @@ class FakeRepository:
     async def get_driver_by_remoteJid(self, remote_jid: str) -> dict[str, Any] | None:
         return self.drivers_by_remote_jid.get(remote_jid)
 
+    async def get_driver_by_phone_number(
+        self,
+        phone_number: str,
+    ) -> dict[str, Any] | None:
+        """Look up a driver by phone_number column.
+
+        Supports multiple phones separated by '/': tries each one until a match is found.
+        """
+        if "/" in phone_number:
+            phones = phone_number.split("/")
+            for phone in phones:
+                result = await self._get_driver_by_single_phone(phone.strip())
+                if result:
+                    return result
+            return None
+        return await self._get_driver_by_single_phone(phone_number)
+
+    async def _get_driver_by_single_phone(
+        self,
+        phone: str,
+    ) -> dict[str, Any] | None:
+        for customer in self.customers_by_remote_jid.values():
+            stored_phone = customer.get("phone_number") or ""
+            # Match exact or within /-separated list
+            if phone == stored_phone or phone in stored_phone.split("/"):
+                for driver in self.drivers_by_remote_jid.values():
+                    if driver.get("customer_id") == customer["id"]:
+                        return driver
+        return None
+
     async def create_driver(self, *, customer_id: str) -> dict[str, Any]:
         customer = next(
             (customer for customer in self.customers_by_remote_jid.values() if customer["id"] == customer_id),

tests/test_group_message_service.py

@@ -415,3 +415,257 @@ async def test_trip_ad_with_missing_name_uses_none(settings: Settings) -> None:
     assert len(repo.created_trips) == 1
     customer = list(repo.customers_by_remote_jid.values())[0]
     assert customer["name"] is None
+
+
+@pytest.mark.asyncio
+async def test_multiple_phone_numbers_are_separated_by_slash(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(
+        response_content=_trip_ad_response(driver_phone="967712345678/967876543210")
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        ai=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/967876543210"
+
+
+@pytest.mark.asyncio
+async def test_unregistered_driver_has_null_remote_jid(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content=_trip_ad_response())
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        ai=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["remoteJid"] is None
+    assert customer["phone_number"] == "967712345678"
+
+
+@pytest.mark.asyncio
+async def test_phone_normalization_with_multiple_numbers(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(
+        response_content=_trip_ad_response(driver_phone="+967-71-234-5678 / +967-78-765-4321")
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        ai=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/967787654321"
+
+
+@pytest.mark.asyncio
+async def test_local_phone_number_gets_country_code_prepended(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(
+        response_content=_trip_ad_response(driver_phone="712345678")
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        ai=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_leading_zero_stripped_then_country_code_prepended(settings: Settings) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(
+        response_content=_trip_ad_response(driver_phone="0712345678")
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        ai=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_existing_unregistered_driver_with_multiple_phones_adds_new_trip(
+    settings: Settings,
+) -> None:
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(
+        response_content=_trip_ad_response(driver_phone="967712345678/967876543210")
+    )
+
+    await repo.upsert_customer(
+        remote_jid=None,
+        name="أحمد",
+        phone_number="967712345678/967876543210",
+        registered=False,
+    )
+
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        ai=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_get_driver_by_phone_number_with_multiple_phones(settings: Settings) -> None:
+    repo = FakeRepository()
+
+    customer = await repo.upsert_customer(
+        remote_jid=None,
+        name="أحمد",
+        phone_number="967712345678/967876543210",
+        registered=False,
+    )
+    await repo.create_driver(customer_id=str(customer["id"]))
+
+    driver = await repo.get_driver_by_phone_number("967712345678/967876543210")
+    assert driver is not None
+    assert driver["customers"]["phone_number"] == "967712345678/967876543210"
+
+
+@pytest.mark.asyncio
+async def test_unregistered_driver_later_dm_merges_into_same_row(settings: Settings) -> None:
+    """When a driver saved from a group (remoteJid=None) later DMs us,
+    the DM should update the existing row instead of creating a duplicate."""
+    repo = FakeRepository()
+    embeddings = FakeEmbeddings()
+    provider = FakeProvider(response_content=_trip_ad_response())
+
+    # Step 1: Driver posts trip ad in group (creates unregistered customer)
+    service = GroupMessageService(
+        repository=repo,
+        embeddings=embeddings,
+        ai=provider,
+        settings=settings,
+    )
+    inbound = _make_inbound(text="رحلة من صنعاء إلى عدن")
+    await service.handle_group_message(inbound)
+
+    assert len(repo.customers_by_remote_jid) == 1
+    existing = list(repo.customers_by_remote_jid.values())[0]
+    assert existing["remoteJid"] is None
+    assert existing["phone_number"] == "967712345678"
+    existing_id = existing["id"]
+
+    # Step 2: Same driver sends a DM (should update existing row, not create new one)
+    customer = await repo.upsert_customer(
+        remote_jid="967712345678",
+        name="أحمد",
+        phone_number="967712345678",
+        registered=True,
+    )
+
+    assert customer["id"] == existing_id
+    assert customer["remoteJid"] == "967712345678"
+    assert customer["registered"] is True
+    # Should still be only one customer row
+    assert len(repo.customers_by_remote_jid) == 1
+
+
+@pytest.mark.asyncio
+async def test_dm_with_shared_phone_merges_into_existing(settings: Settings) -> None:
+    """When a customer exists with phone_number matching a DM's phone,
+    the DM should update the existing row."""
+    repo = FakeRepository()
+
+    # Create a customer via DM first
+    customer1 = await repo.upsert_customer(
+        remote_jid="967712345678",
+        name="أحمد",
+        phone_number="967712345678",
+        registered=True,
+    )
+
+    # Another DM with same phone_number but different remoteJid (e.g. new device)
+    customer2 = await repo.upsert_customer(
+        remote_jid="967999999999",
+        name="أحمد",
+        phone_number="967712345678",
+        registered=True,
+    )
+
+    assert customer1["id"] == customer2["id"]
+    assert len(repo.customers_by_remote_jid) == 1
+
+
+@pytest.mark.asyncio
+async def test_dm_with_multi_phone_merges_into_existing(settings: Settings) -> None:
+    """When a DM's phone_number matches one of the phones in an existing
+    /-separated phone_number, it should update the existing row."""
+    repo = FakeRepository()
+
+    # Create unregistered driver with two phones
+    customer1 = await repo.upsert_customer(
+        remote_jid=None,
+        name="أحمد",
+        phone_number="967712345678/967876543210",
+        registered=False,
+    )
+
+    # DM comes in with just the first phone number
+    customer2 = await repo.upsert_customer(
+        remote_jid="967712345678",
+        name="أحمد",
+        phone_number="967712345678",
+        registered=True,
+    )
+
+    assert customer1["id"] == customer2["id"]
+    assert customer2["remoteJid"] == "967712345678"
+    assert customer2["registered"] is True