feat(segments): P1/4 — the segment table, create and read a static list

app/crm/api.py

@@ -41,6 +41,9 @@
 router.include_router(
     outreach_api.customer_router, prefix="/customers", tags=["Workflows"]
 )
+router.include_router(
+    outreach_api.segment_router, prefix="/segments", tags=["Segments"]
+)
 router.include_router(
     connectivity_api.router, prefix="/connectors", tags=["Connectors"]
 )

app/crm/outreach/api.py

@@ -24,7 +24,7 @@
 
 from app.api.security.breeze_buddy.rbac_token import get_current_user_with_rbac
 from app.crm.auth import merchant_scope
-from app.crm.outreach import analytics, plans, runs, versions
+from app.crm.outreach import analytics, plans, runs, segments, versions
 from app.crm.outreach.schemas import (
     CustomerRun,
     EnrollmentRun,
@@ -33,6 +33,8 @@
     RunCall,
     RunPage,
     RunStep,
+    Segment,
+    SegmentPage,
     VersionMigration,
     Workflow,
     WorkflowCallSummary,
@@ -48,13 +50,22 @@
 # The customer-facing door (/customers/{id}/runs) mounts under a different
 # prefix than the plans' router — the record module's two-router precedent.
 customer_router = APIRouter()
+# The audiences door (/segments, canon T15) — outreach owns segments beside
+# the plans they will one day feed through a broadcast; same precedent.
+segment_router = APIRouter()
 
 
 class WorkflowCreate(BaseModel):
     name: str = Field(min_length=1, max_length=200)
     definition: Dict[str, Any]
 
 
+class SegmentCreate(BaseModel):
+    # Phase 1 creates static lists only: a name and nothing else. A
+    # dynamic segment's definition joins this body with its compiler.
+    name: str = Field(min_length=1, max_length=200)
+
+
 class WorkflowStatusChange(BaseModel):
     status: Literal["live", "paused", "archived"]
 
@@ -470,3 +481,50 @@ async def customer_runs_route(
     ),
 ) -> List[CustomerRun]:
     return await runs.customer_runs(merchant_id, customer_id, limit)
+
+
+@segment_router.post("", response_model=Segment, status_code=status.HTTP_201_CREATED)
+async def create_segment_route(
+    body: SegmentCreate,
+    merchant_id: str = Depends(
+        merchant_scope("create a segment", "crm.segments.create")
+    ),
+    current_user: UserInfo = Depends(get_current_user_with_rbac),
+) -> Segment:
+    """A new, empty static list. 409 when the merchant already has a segment
+    by this name; 422 when the name is blank once trimmed."""
+    try:
+        return await segments.create_static_list(
+            merchant_id, body.name, current_user.email
+        )
+    except segments.SegmentNameTaken as e:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"a segment named {e} already exists",
+        )
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e)
+        )
+
+
+@segment_router.get("", response_model=SegmentPage)
+async def list_segments_route(
+    limit: int = Query(50, ge=1, le=200),
+    offset: int = Query(0, ge=0),
+    merchant_id: str = Depends(merchant_scope("list segments", "crm.segments.list")),
+) -> SegmentPage:
+    return await segments.list_segments(merchant_id, limit, offset)
+
+
+@segment_router.get("/{segment_id}", response_model=Segment)
+async def get_segment_route(
+    segment_id: UUID,
+    merchant_id: str = Depends(merchant_scope("read a segment", "crm.segments.get")),
+) -> Segment:
+    segment = await segments.get_segment(merchant_id, str(segment_id))
+    if segment is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Segment not found"
+        )
+    return segment

app/crm/outreach/db/accessors/segment.py

@@ -0,0 +1,48 @@
+"""Mechanical DB access for crm_segment (T15) — one table, one file (module
+rules §1 at scale). Every function self-scopes one statement; no decisions.
+"""
+
+from typing import List, Optional
+
+from app.crm.outreach.db.decoders.segment import decode_segment
+from app.crm.outreach.db.queries.segment import (
+    count_segments_query,
+    get_segment_query,
+    insert_static_list_query,
+    list_segments_query,
+)
+from app.crm.outreach.schemas import Segment
+from app.crm.shared.db import crm_connection
+
+
+async def insert_static_list(
+    merchant_id: str, name: str, created_by: Optional[str]
+) -> Segment:
+    """Raises the db door's UniqueViolation when the merchant already has a
+    segment with this name — the caller says what that means."""
+    query, values = insert_static_list_query(merchant_id, name, created_by)
+    async with crm_connection() as conn:
+        row = await conn.fetchrow(query, *values)
+    assert row is not None  # INSERT ... RETURNING always yields the row
+    return decode_segment(row)
+
+
+async def get_segment(merchant_id: str, segment_id: str) -> Optional[Segment]:
+    query, values = get_segment_query(merchant_id, segment_id)
+    async with crm_connection() as conn:
+        row = await conn.fetchrow(query, *values)
+    return decode_segment(row) if row else None
+
+
+async def list_segments(merchant_id: str, limit: int, offset: int) -> List[Segment]:
+    query, values = list_segments_query(merchant_id, limit, offset)
+    async with crm_connection() as conn:
+        rows = await conn.fetch(query, *values)
+    return [decode_segment(row) for row in rows]
+
+
+async def count_segments(merchant_id: str) -> int:
+    query, values = count_segments_query(merchant_id)
+    async with crm_connection() as conn:
+        row = await conn.fetchrow(query, *values)
+    return int(row["total"]) if row else 0

app/crm/outreach/db/decoders/segment.py

@@ -0,0 +1,27 @@
+"""row -> schema translation for crm_segment (T15) — one table, one file
+(module rules §1 at scale). DB-side translation only — never imported
+outside db/.
+"""
+
+from typing import Any, Mapping
+
+from app.crm.outreach.schemas import Segment
+from app.crm.shared.decode import jsonb_value
+
+
+def decode_segment(row: Mapping[str, Any]) -> Segment:
+    definition = jsonb_value(row["definition"])
+    return Segment(
+        id=row["id"],
+        merchant_id=row["merchant_id"],
+        name=row["name"],
+        # The kind is a predicate over definition, never a stored column
+        # (law 10): NULL = a static list whose members are rows.
+        kind="static" if definition is None else "dynamic",
+        definition=definition,
+        member_count=row["member_count"],
+        computed_at=row["computed_at"],
+        created_by=row["created_by"],
+        created_at=row["created_at"],
+        updated_at=row["updated_at"],
+    )

app/crm/outreach/db/queries/segment.py

@@ -0,0 +1,64 @@
+"""SQL builders for crm_segment (T15, canon/05-audiences.md) — one table, one
+file (module rules §1 at scale). $1 placeholders only — every value
+parameterized; merchant_id is $1 on every read and write (tenancy law).
+"""
+
+from typing import Any, List, Optional, Tuple
+
+from app.crm.outreach.db.queries.tables import SEGMENT_TABLE
+
+_SEGMENT_COLUMNS = """
+    id, merchant_id, name, definition, member_count, computed_at,
+    created_by, created_at, updated_at
+"""
+
+
+def insert_static_list_query(
+    merchant_id: str, name: str, created_by: Optional[str]
+) -> Tuple[str, List[Any]]:
+    """A static list is born empty, and its count is exact from the first
+    moment: zero members, stamped now (T15 col 7 — exact forever for a
+    list). ``definition`` stays NULL — that NULL is what makes it a list.
+    No ON CONFLICT: a second list with the same name is the unique index's
+    refusal to surface, never a merge."""
+    query = f"""
+        INSERT INTO {SEGMENT_TABLE}
+            (merchant_id, name, member_count, computed_at, created_by)
+        VALUES ($1, $2, 0, now(), $3)
+        RETURNING {_SEGMENT_COLUMNS}
+    """
+    return query, [merchant_id, name, created_by]
+
+
+def get_segment_query(merchant_id: str, segment_id: str) -> Tuple[str, List[Any]]:
+    query = f"""
+        SELECT {_SEGMENT_COLUMNS}
+        FROM {SEGMENT_TABLE}
+        WHERE merchant_id = $1 AND id = $2
+    """
+    return query, [merchant_id, segment_id]
+
+
+def list_segments_query(
+    merchant_id: str, limit: int, offset: int
+) -> Tuple[str, List[Any]]:
+    """Newest first — the order crm_segment_merchant_created_ix carries."""
+    query = f"""
+        SELECT {_SEGMENT_COLUMNS}
+        FROM {SEGMENT_TABLE}
+        WHERE merchant_id = $1
+        ORDER BY created_at DESC, id DESC
+        LIMIT $2 OFFSET $3
+    """
+    return query, [merchant_id, limit, offset]
+
+
+def count_segments_query(merchant_id: str) -> Tuple[str, List[Any]]:
+    """The merchant's segment count, for the list's pager (in the body —
+    the RunPage ruling)."""
+    query = f"""
+        SELECT count(*)::int AS total
+        FROM {SEGMENT_TABLE}
+        WHERE merchant_id = $1
+    """
+    return query, [merchant_id]

app/crm/outreach/db/queries/tables.py

@@ -8,3 +8,4 @@
 ENROLLMENT_TABLE = "crm_workflow_enrollment"
 VERSION_TABLE = "crm_workflow_version"
 STEP_TABLE = "crm_workflow_step"
+SEGMENT_TABLE = "crm_segment"

app/crm/outreach/schemas.py

@@ -1032,3 +1032,34 @@ class CustomerRun(EnrollmentRun):
     each row says which plan it belongs to (rollout phase 09)."""
 
     workflow_name: str
+
+
+# --- segments (canon T15, canon/05-audiences.md) ---
+
+
+class Segment(BaseModel):
+    """A segment: a static list (``definition`` None — its members are
+    stored rows) or a dynamic segment (``definition`` set — the stored
+    question, computed at read time). ``kind`` is derived from
+    ``definition`` on read, never stored (law 10). ``member_count`` is
+    display-only and always travels with ``computed_at``, so a count says
+    how old it is (T15 cols 7-8); nothing ever decides on it."""
+
+    id: UUID
+    merchant_id: str
+    name: str
+    kind: Literal["static", "dynamic"]
+    definition: Optional[Dict[str, Any]] = None
+    member_count: Optional[int] = None
+    computed_at: Optional[datetime] = None
+    created_by: Optional[str] = None
+    created_at: datetime
+    updated_at: datetime
+
+
+class SegmentPage(BaseModel):
+    """One page of a merchant's segments and how many there are in all —
+    the total in the body, typed (the RunPage ruling)."""
+
+    items: List[Segment]
+    total: int

app/crm/outreach/segments.py

@@ -0,0 +1,67 @@
+"""Segments (canon T15, canon/05-audiences.md) — phase 1: a static list's
+row, created and read.
+
+A segment is a stored question; membership is computed at read time and
+never stored. A static list is the one exception — an imported set IS its
+members (phase 2 adds them; phase 4 compiles dynamic definitions). Until
+then every segment this file creates is a static list.
+
+What this file keeps:
+- **A segment is never a permission list** (modules/05-outreach.md:233).
+  Nothing here sends, consents or suppresses; a broadcast still asks the
+  gate about every member.
+- **The name is unique per merchant** (T15 col 3). The unique index referees
+  two racing creates; this file only names the refusal.
+- **Counts carry their stamp** (T15 col 8): a new list's count is exact —
+  zero members, stamped at birth.
+"""
+
+from typing import Optional
+
+from app.crm.outreach.db import UniqueViolation
+from app.crm.outreach.db.accessors import segment as segment_accessor
+from app.crm.outreach.schemas import Segment, SegmentPage
+
+
+class SegmentNameTaken(Exception):
+    """The merchant already has a segment with this name (T15 col 3)."""
+
+
+def segment_name(raw: str) -> str:
+    """The name as stored: surrounding whitespace dropped, so "Diwali VIPs"
+    and "Diwali VIPs " are one name to the unique index, not two lists a
+    merchant cannot tell apart. Pure. Raises ValueError when nothing is
+    left — a blank name is not a name. The 085 CHECK refuses a blank or
+    untrimmed name too, so a writer that skips this cannot split a name."""
+    name = (raw or "").strip()
+    if not name:
+        raise ValueError("a segment needs a name")
+    return name
+
+
+async def create_static_list(
+    merchant_id: str, name: str, created_by: Optional[str]
+) -> Segment:
+    """A new, empty static list. Raises SegmentNameTaken when the merchant
+    already has a segment by this name, ValueError on a blank name."""
+    stored = segment_name(name)
+    try:
+        return await segment_accessor.insert_static_list(
+            merchant_id, stored, created_by
+        )
+    except UniqueViolation as e:
+        raise SegmentNameTaken(stored) from e
+
+
+async def get_segment(merchant_id: str, segment_id: str) -> Optional[Segment]:
+    """One segment, or None when this merchant has no such segment — another
+    merchant's id answers None too, never their row."""
+    return await segment_accessor.get_segment(merchant_id, segment_id)
+
+
+async def list_segments(merchant_id: str, limit: int, offset: int) -> SegmentPage:
+    """One page of the merchant's segments, newest first, and the total."""
+    return SegmentPage(
+        items=await segment_accessor.list_segments(merchant_id, limit, offset),
+        total=await segment_accessor.count_segments(merchant_id),
+    )