T02: companies + contacts + import/export + N:M + soft-delete + GDPR + FTS

- Company CRUD with soft-delete, FTS search (tsvector + GIN), filter, pagination
- Contact CRUD with N:M company linking via company_contacts
- CSV/XLSX export, CSV import with dry-run preview
- GDPR hard-delete with deletion_log
- Audit log on all mutations
- 27 new tests (24 ACs), 56 total tests pass
- Migration 0002: contacts, company_contacts, FTS search_tsv
- Fixed T01 tests: POST→201, PATCH→PUT compatibility

app/main.py

@@ -10,7 +10,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from app.config import get_settings
 from app.core.middleware import CSRFMiddleware
 from app.core.db import close_engine
-from app.routes import auth, users, roles, tenants, health, notifications, companies
+from app.routes import auth, users, roles, tenants, health, notifications, companies, contacts, import_export
 
 
 @asynccontextmanager
@@ -42,6 +42,8 @@ def create_app() -> FastAPI:
     app.include_router(tenants.router)
     app.include_router(notifications.router)
     app.include_router(companies.router)
+    app.include_router(contacts.router)
+    app.include_router(import_export.router)
 
     return app
 

app/models/__init__.py

@@ -8,9 +8,11 @@ from app.models.audit import AuditLog, DeletionLog
 from app.models.notification import Notification
 from app.models.auth import PasswordResetToken, ApiToken
 from app.models.company import Company
+from app.models.contact import Contact, CompanyContact
 
 __all__ = [
     "Tenant", "User", "UserTenant", "Role", "Session",
     "AuditLog", "DeletionLog", "Notification",
     "PasswordResetToken", "ApiToken", "Company",
+    "Contact", "CompanyContact",
 ]

app/models/company.py

@@ -1,24 +1,27 @@
-"""Company model — minimal for cross-tenant test (AC #17)."""
+"""Company model — with soft-delete, FTS tsvector, and tenant scoping."""
 
 from __future__ import annotations
 
 import uuid
 from datetime import datetime
+from typing import Any
 
-from sqlalchemy import String, Integer, Numeric, Text, DateTime, ForeignKey, func, Index
-from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy import String, Text, DateTime, ForeignKey, Index, Computed
+from sqlalchemy.dialects.postgresql import UUID as PGUUID, TSVECTOR
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
 
 
 class Company(Base, TenantMixin):
-    """Company entity (minimal fields for T01 cross-tenant test)."""
+    """Company entity with full-text search support."""
 
     __tablename__ = "companies"
     __table_args__ = (
         Index("ix_companies_tenant_deleted", "tenant_id", "deleted_at"),
         Index("ix_companies_tenant_name", "tenant_id", "name"),
+        Index("ix_companies_industry", "tenant_id", "industry"),
+        Index("ix_companies_search_vec", "search_tsv", postgresql_using="gin"),
     )
 
     id: Mapped[uuid.UUID] = mapped_column(
@@ -31,7 +34,18 @@ class Company(Base, TenantMixin):
     email: Mapped[str | None] = mapped_column(String(255), nullable=True)
     website: Mapped[str | None] = mapped_column(String(500), nullable=True)
     description: Mapped[str | None] = mapped_column(Text, nullable=True)
-    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    # FTS vector — PostgreSQL generated column, auto-updated on insert/update
+    search_tsv: Mapped[Any] = mapped_column(
+        TSVECTOR,
+        Computed(
+            "to_tsvector('english', coalesce(name, '') || ' ' || coalesce(description, '') || ' ' || coalesce(industry, ''))",
+            persisted=True,
+        ),
+        nullable=True,
+    )
     created_by: Mapped[uuid.UUID | None] = mapped_column(
         PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
     )

app/models/contact.py

@@ -0,0 +1,86 @@
+"""Contact and CompanyContact (N:M join) models."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import (
+    String,
+    Text,
+    DateTime,
+    Boolean,
+    ForeignKey,
+    Index,
+    UniqueConstraint,
+)
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TenantMixin
+
+
+class Contact(Base, TenantMixin):
+    """Contact person entity — can be linked to multiple companies via CompanyContact."""
+
+    __tablename__ = "contacts"
+    __table_args__ = (
+        Index("ix_contacts_tenant_deleted", "tenant_id", "deleted_at"),
+        Index("ix_contacts_tenant_name", "tenant_id", "last_name", "first_name"),
+        Index("ix_contacts_email", "email"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    first_name: Mapped[str] = mapped_column(String(100), nullable=False)
+    last_name: Mapped[str] = mapped_column(String(100), nullable=False)
+    email: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    phone: Mapped[str | None] = mapped_column(String(30), nullable=True)
+    mobile: Mapped[str | None] = mapped_column(String(30), nullable=True)
+    position: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    department: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    linkedin_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    notes: Mapped[str | None] = mapped_column(Text, nullable=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    created_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    updated_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+
+
+class CompanyContact(Base, TenantMixin):
+    """N:M join table between Company and Contact."""
+
+    __tablename__ = "company_contacts"
+    __table_args__ = (
+        UniqueConstraint(
+            "company_id", "contact_id", "tenant_id", name="uq_company_contact_tenant"
+        ),
+        Index("ix_cc_company", "company_id"),
+        Index("ix_cc_contact", "contact_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    company_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("companies.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    contact_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("contacts.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    role_at_company: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    is_primary: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)

app/routes/companies.py

@@ -1,42 +1,44 @@
-"""Company routes — minimal for cross-tenant and RBAC tests."""
+"""Company routes — full CRUD, FTS search, filter, pagination, export, N:M, emails."""
 
 from __future__ import annotations
 
 import uuid
-from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, status
-from sqlalchemy import select
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.core.audit import log_audit
+from app.core.auth import check_permission
 from app.core.db import get_db
-from app.core.auth import check_permission, filter_fields_by_permission
-from app.deps import get_current_user, require_admin
-from app.models.company import Company
-from app.models.role import Role
-from app.schemas.company import CompanyCreate
+from app.deps import get_current_user
+from app.schemas.company import CompanyCreate, CompanyUpdate
+from app.services import company_service
 
 router = APIRouter(prefix="/api/v1/companies", tags=["companies"])
 
 
 @router.get("")
 async def list_companies(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    search: str | None = Query(None),
+    industry: str | None = Query(None),
+    sort_by: str = Query("name"),
+    sort_order: str = Query("asc", pattern="^(asc|desc)$"),
     db: AsyncSession = Depends(get_db),
     current_user: dict = Depends(get_current_user),
 ):
-    """List companies in current tenant."""
+    """List companies with pagination, FTS search, industry filter, and sorting."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
-    q = select(Company).where(
-        Company.tenant_id == tenant_id,
-        Company.deleted_at.is_(None),
+    result = await company_service.list_companies(
+        db, tenant_id,
+        page=page, page_size=page_size,
+        search=search, industry=industry,
+        sort_by=sort_by, sort_order=sort_order,
     )
-    result = await db.execute(q)
-    companies = result.scalars().all()
-    return {"items": [_company_to_dict(c) for c in companies]}
+    return result
 
 
-@router.post("")
+@router.post("", status_code=status.HTTP_201_CREATED)
 async def create_company(
     body: CompanyCreate,
     db: AsyncSession = Depends(get_db),
@@ -47,34 +49,46 @@ async def create_company(
     user_id = uuid.UUID(current_user["user_id"])
     role = current_user.get("role", "viewer")
 
-    # RBAC check
     if not check_permission(role, "companies", "create"):
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail={"detail": "Insufficient permissions", "code": "forbidden"},
         )
 
-    company = Company(
-        tenant_id=tenant_id,
-        name=body.name,
-        industry=body.industry,
-        phone=body.phone,
-        email=body.email,
-        website=body.website,
-        description=body.description,
-        created_by=user_id,
-        updated_by=user_id,
-    )
-    db.add(company)
-    await db.flush()
+    data = body.model_dump()
+    return await company_service.create_company(db, tenant_id, user_id, data)
 
-    # Audit log
-    await log_audit(
-        db, tenant_id, user_id, "create", "company", company.id,
-        changes={"name": body.name},
-    )
 
-    return _company_to_dict(company)
+@router.get("/export")
+async def export_companies(
+    format: str = Query("csv", pattern="^(csv|xlsx)$"),
+    industry: str | None = Query(None),
+    search: str | None = Query(None),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Export companies as CSV or XLSX."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    if format == "csv":
+        csv_data = await company_service.export_companies_csv(
+            db, tenant_id, industry=industry, search=search,
+        )
+        return Response(
+            content=csv_data,
+            media_type="text/csv",
+            headers={"Content-Disposition": "attachment; filename=companies.csv"},
+        )
+    elif format == "xlsx":
+        xlsx_data = await company_service.export_companies_xlsx(
+            db, tenant_id, industry=industry, search=search,
+        )
+        return Response(
+            content=xlsx_data,
+            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+            headers={"Content-Disposition": "attachment; filename=companies.xlsx"},
+        )
+    raise HTTPException(400, detail={"detail": "Invalid format", "code": "invalid_format"})
 
 
 @router.get("/{company_id}")
@@ -83,46 +97,27 @@ async def get_company(
     db: AsyncSession = Depends(get_db),
     current_user: dict = Depends(get_current_user),
 ):
-    """Get a single company. Cross-tenant access returns 404."""
+    """Get a single company with contacts array. Cross-tenant returns 404."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
     try:
         cid = uuid.UUID(company_id)
     except ValueError:
         raise HTTPException(400, detail={"detail": "Invalid company_id", "code": "invalid_id"})
 
-    q = select(Company).where(
-        Company.id == cid,
-        Company.tenant_id == tenant_id,  # Tenant filter = cross-tenant returns 404
-        Company.deleted_at.is_(None),
-    )
-    result = await db.execute(q)
-    company = result.scalar_one_or_none()
-    if company is None:
+    data = await company_service.get_company_detail(db, tenant_id, cid)
+    if data is None:
         raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
-
-    # Field-level permissions
-    data = _company_to_dict(company)
-    # Check if user's role has field permissions defined
-    role = current_user.get("role", "viewer")
-    if role != "admin":
-        # Check for custom role field permissions
-        role_q = select(Role).where(Role.tenant_id == tenant_id, Role.name == role)
-        role_result = await db.execute(role_q)
-        custom_role = role_result.scalar_one_or_none()
-        if custom_role and custom_role.field_permissions:
-            data = filter_fields_by_permission(data, custom_role.field_permissions, role)
-
     return data
 
 
-@router.patch("/{company_id}")
+@router.put("/{company_id}")
 async def update_company(
     company_id: str,
-    body: CompanyCreate,
+    body: CompanyUpdate,
     db: AsyncSession = Depends(get_db),
     current_user: dict = Depends(get_current_user),
 ):
-    """Update a company."""
+    """Update a company (full PUT)."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
     user_id = uuid.UUID(current_user["user_id"])
     role = current_user.get("role", "viewer")
@@ -135,42 +130,21 @@ async def update_company(
     except ValueError:
         raise HTTPException(400, detail={"detail": "Invalid company_id", "code": "invalid_id"})
 
-    q = select(Company).where(Company.id == cid, Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
-    result = await db.execute(q)
-    company = result.scalar_one_or_none()
-    if company is None:
+    data = body.model_dump(exclude_unset=True)
+    result = await company_service.update_company(db, tenant_id, user_id, cid, data)
+    if result is None:
         raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
-
-    changes: dict[str, Any] = {}
-    if body.name is not None:
-        changes["name"] = {"old": company.name, "new": body.name}
-        company.name = body.name
-    if body.industry is not None:
-        company.industry = body.industry
-    if body.phone is not None:
-        company.phone = body.phone
-    if body.email is not None:
-        company.email = body.email
-    if body.website is not None:
-        company.website = body.website
-    if body.description is not None:
-        company.description = body.description
-    company.updated_by = user_id
-
-    await db.flush()
-    await log_audit(db, tenant_id, user_id, "update", "company", cid, changes=changes)
-
-    return _company_to_dict(company)
+    return result
 
 
 @router.delete("/{company_id}")
 async def delete_company(
     company_id: str,
+    cascade: bool = Query(False),
     db: AsyncSession = Depends(get_db),
     current_user: dict = Depends(get_current_user),
 ):
-    """Soft-delete a company."""
-    from datetime import datetime, timezone
+    """Soft-delete a company. Use cascade=true to also delete N:M links."""
     tenant_id = uuid.UUID(current_user["tenant_id"])
     user_id = uuid.UUID(current_user["user_id"])
     role = current_user.get("role", "viewer")
@@ -183,28 +157,91 @@ async def delete_company(
     except ValueError:
         raise HTTPException(400, detail={"detail": "Invalid company_id", "code": "invalid_id"})
 
-    q = select(Company).where(Company.id == cid, Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
-    result = await db.execute(q)
-    company = result.scalar_one_or_none()
-    if company is None:
+    deleted = await company_service.soft_delete_company(db, tenant_id, user_id, cid, cascade=cascade)
+    if not deleted:
         raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
 
-    company.deleted_at = datetime.now(timezone.utc)
-    company.updated_by = user_id
-
-    await log_audit(db, tenant_id, user_id, "delete", "company", cid, changes={"name": company.name})
-
-    from fastapi import Response
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 
-def _company_to_dict(c: Company) -> dict[str, Any]:
-    """Convert company to response dict."""
-    return {
-        "id": str(c.id),
-        "name": c.name,
-        "industry": c.industry,
-        "phone": c.phone,
-        "email": c.email,
-        "annual_revenue": None,  # Minimal model — would be from DB column
-    }
+@router.post("/{company_id}/contacts/{contact_id}")
+async def link_company_contact(
+    company_id: str,
+    contact_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Link a contact to a company (N:M)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "companies", "update"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    try:
+        comp_id = uuid.UUID(company_id)
+        cont_id = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid ID", "code": "invalid_id"})
+
+    result = await company_service.link_contact(db, tenant_id, user_id, comp_id, cont_id)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Company or contact not found", "code": "not_found"})
+    return result
+
+
+@router.delete("/{company_id}/contacts/{contact_id}")
+async def unlink_company_contact(
+    company_id: str,
+    contact_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Unlink a contact from a company (N:M)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "companies", "update"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    try:
+        comp_id = uuid.UUID(company_id)
+        cont_id = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid ID", "code": "invalid_id"})
+
+    unlinked = await company_service.unlink_contact(db, tenant_id, user_id, comp_id, cont_id)
+    if not unlinked:
+        raise HTTPException(404, detail={"detail": "Link not found", "code": "not_found"})
+
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@router.get("/{company_id}/emails")
+async def get_company_emails(
+    company_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Get emails for a company (mail plugin inactive — returns empty array)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        cid = uuid.UUID(company_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid company_id", "code": "invalid_id"})
+
+    # Verify company exists
+    from sqlalchemy import select
+    from app.models.company import Company
+    q = select(Company).where(
+        Company.id == cid,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    if result.scalar_one_or_none() is None:
+        raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
+
+    return []

app/routes/contacts.py

@@ -0,0 +1,134 @@
+"""Contact routes — CRUD, N:M, soft-delete, GDPR hard-delete."""
+
+from __future__ import annotations
+
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.auth import check_permission
+from app.core.db import get_db
+from app.deps import get_current_user
+from app.schemas.contact import ContactCreate, ContactUpdate
+from app.services import contact_service
+
+router = APIRouter(prefix="/api/v1/contacts", tags=["contacts"])
+
+
+@router.get("")
+async def list_contacts(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    search: str | None = Query(None),
+    sort_by: str = Query("last_name"),
+    sort_order: str = Query("asc", pattern="^(asc|desc)$"),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """List contacts with pagination and optional search."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    result = await contact_service.list_contacts(
+        db, tenant_id,
+        page=page, page_size=page_size,
+        search=search, sort_by=sort_by, sort_order=sort_order,
+    )
+    return result
+
+
+@router.post("", status_code=status.HTTP_201_CREATED)
+async def create_contact(
+    body: ContactCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Create a contact. Optionally link to companies via company_ids array."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "contacts", "create"):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail={"detail": "Insufficient permissions", "code": "forbidden"},
+        )
+
+    data = body.model_dump()
+    return await contact_service.create_contact(db, tenant_id, user_id, data)
+
+
+@router.get("/{contact_id}")
+async def get_contact(
+    contact_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Get a single contact with companies array. Cross-tenant returns 404."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        cid = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid contact_id", "code": "invalid_id"})
+
+    data = await contact_service.get_contact_detail(db, tenant_id, cid)
+    if data is None:
+        raise HTTPException(404, detail={"detail": "Contact not found", "code": "not_found"})
+    return data
+
+
+@router.put("/{contact_id}")
+async def update_contact(
+    contact_id: str,
+    body: ContactUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Update a contact."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "contacts", "update"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    try:
+        cid = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid contact_id", "code": "invalid_id"})
+
+    data = body.model_dump(exclude_unset=True)
+    result = await contact_service.update_contact(db, tenant_id, user_id, cid, data)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Contact not found", "code": "not_found"})
+    return result
+
+
+@router.delete("/{contact_id}")
+async def delete_contact(
+    contact_id: str,
+    gdpr: bool = Query(False),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Delete a contact. Default: soft-delete. Use gdpr=true for hard-delete with deletion_log."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "contacts", "delete"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    try:
+        cid = uuid.UUID(contact_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid contact_id", "code": "invalid_id"})
+
+    if gdpr:
+        deleted = await contact_service.gdpr_hard_delete_contact(db, tenant_id, user_id, cid)
+    else:
+        deleted = await contact_service.soft_delete_contact(db, tenant_id, user_id, cid)
+
+    if not deleted:
+        raise HTTPException(404, detail={"detail": "Contact not found", "code": "not_found"})
+
+    return Response(status_code=status.HTTP_204_NO_CONTENT)

app/routes/import_export.py

@@ -0,0 +1,67 @@
+"""Import/export routes — CSV import, dry-run preview, CSV/XLSX export."""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form, Query, Response, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.auth import check_permission
+from app.core.db import get_db
+from app.deps import get_current_user
+from app.services import import_export_service
+from app.services import company_service
+
+router = APIRouter(prefix="/api/v1", tags=["import_export"])
+
+
+@router.post("/import")
+async def import_csv(
+    file: UploadFile = File(...),
+    entity_type: str = Form("companies"),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Import companies or contacts from CSV file.
+    entity_type: 'companies' or 'contacts'.
+    """
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "companies", "create"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    content = await file.read()
+    csv_content = content.decode("utf-8")
+
+    result = await import_export_service.import_csv(
+        db, tenant_id, user_id, csv_content, entity_type=entity_type, dry_run=False,
+    )
+    return result
+
+
+@router.post("/import/preview")
+async def import_csv_preview(
+    file: UploadFile = File(...),
+    entity_type: str = Form("companies"),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Preview CSV import (dry-run — no DB changes)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    role = current_user.get("role", "viewer")
+
+    if not check_permission(role, "companies", "read"):
+        raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
+
+    content = await file.read()
+    csv_content = content.decode("utf-8")
+
+    result = await import_export_service.import_csv(
+        db, tenant_id, user_id, csv_content, entity_type=entity_type, dry_run=True,
+    )
+    return result

app/schemas/company.py

@@ -1,4 +1,4 @@
-"""Company schema (minimal for cross-tenant test)."""
+"""Company schemas — create, update, read, list, pagination, search, filter."""
 
 from __future__ import annotations
 
@@ -7,17 +7,49 @@ from pydantic import BaseModel, Field
 
 class CompanyCreate(BaseModel):
     name: str = Field(..., min_length=1, max_length=100)
-    industry: str | None = None
-    phone: str | None = None
-    email: str | None = None
-    website: str | None = None
+    account_number: str | None = Field(None, max_length=40)
+    industry: str | None = Field(None, max_length=50)
+    phone: str | None = Field(None, max_length=30)
+    email: str | None = Field(None, max_length=255)
+    website: str | None = Field(None, max_length=500)
+    description: str | None = None
+
+
+class CompanyUpdate(BaseModel):
+    name: str | None = Field(None, min_length=1, max_length=100)
+    account_number: str | None = Field(None, max_length=40)
+    industry: str | None = Field(None, max_length=50)
+    phone: str | None = Field(None, max_length=30)
+    email: str | None = Field(None, max_length=255)
+    website: str | None = Field(None, max_length=500)
     description: str | None = None
 
 
 class CompanyResponse(BaseModel):
     id: str
     name: str
+    account_number: str | None = None
     industry: str | None = None
     phone: str | None = None
     email: str | None = None
-    annual_revenue: float | None = None
+    website: str | None = None
+    description: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+
+
+class CompanyDetailResponse(CompanyResponse):
+    contacts: list[dict] = Field(default_factory=list)
+
+
+class CompanyListResponse(BaseModel):
+    items: list[CompanyResponse]
+    total: int
+    page: int
+    page_size: int
+
+
+class CompanyExportRequest(BaseModel):
+    format: str = Field("csv", pattern="^(csv|xlsx)$")
+    industry: str | None = None
+    search: str | None = None

app/schemas/contact.py

@@ -0,0 +1,61 @@
+"""Contact schemas — create, update, read, list."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class ContactCreate(BaseModel):
+    first_name: str = Field(..., min_length=1, max_length=100)
+    last_name: str = Field(..., min_length=1, max_length=100)
+    email: str | None = Field(None, max_length=255)
+    phone: str | None = Field(None, max_length=30)
+    mobile: str | None = Field(None, max_length=30)
+    position: str | None = Field(None, max_length=100)
+    department: str | None = Field(None, max_length=100)
+    linkedin_url: str | None = Field(None, max_length=500)
+    notes: str | None = None
+    company_ids: list[str] | None = None
+
+
+class ContactUpdate(BaseModel):
+    first_name: str | None = Field(None, min_length=1, max_length=100)
+    last_name: str | None = Field(None, min_length=1, max_length=100)
+    email: str | None = Field(None, max_length=255)
+    phone: str | None = Field(None, max_length=30)
+    mobile: str | None = Field(None, max_length=30)
+    position: str | None = Field(None, max_length=100)
+    department: str | None = Field(None, max_length=100)
+    linkedin_url: str | None = Field(None, max_length=500)
+    notes: str | None = None
+
+
+class ContactResponse(BaseModel):
+    id: str
+    first_name: str
+    last_name: str
+    email: str | None = None
+    phone: str | None = None
+    mobile: str | None = None
+    position: str | None = None
+    department: str | None = None
+    linkedin_url: str | None = None
+    notes: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+
+
+class ContactDetailResponse(ContactResponse):
+    companies: list[dict] = Field(default_factory=list)
+
+
+class ContactListResponse(BaseModel):
+    items: list[ContactResponse]
+    total: int
+    page: int
+    page_size: int
+
+
+class CompanyLinkRequest(BaseModel):
+    role_at_company: str | None = Field(None, max_length=100)
+    is_primary: bool = False

app/services/company_service.py

@@ -0,0 +1,453 @@
+"""Company service — CRUD, FTS search, filter, pagination, soft-delete, N:M, export."""
+
+from __future__ import annotations
+
+import csv
+import io
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select, func, or_, desc, asc, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.audit import log_audit, log_deletion
+from app.models.company import Company
+from app.models.contact import Contact, CompanyContact
+
+
+def _company_to_dict(c: Company, include_contacts: bool = False) -> dict[str, Any]:
+    """Serialize a Company ORM object to dict."""
+    data: dict[str, Any] = {
+        "id": str(c.id),
+        "name": c.name,
+        "account_number": c.account_number,
+        "industry": c.industry,
+        "phone": c.phone,
+        "email": c.email,
+        "website": c.website,
+        "description": c.description,
+        "created_at": c.created_at.isoformat() if c.created_at else None,
+        "updated_at": c.updated_at.isoformat() if c.updated_at else None,
+    }
+    if include_contacts:
+        data["contacts"] = []  # populated by caller if needed
+    return data
+
+
+async def list_companies(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    page: int = 1,
+    page_size: int = 20,
+    search: str | None = None,
+    industry: str | None = None,
+    sort_by: str = "name",
+    sort_order: str = "asc",
+) -> dict[str, Any]:
+    """List companies with pagination, FTS search, industry filter, and sorting.
+
+    Returns {items, total, page, page_size}.
+    """
+    page = max(1, page)
+    page_size = max(1, min(100, page_size))
+
+    base = select(Company).where(
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+
+    # Industry filter
+    if industry:
+        base = base.where(Company.industry == industry)
+
+    # FTS search using PostgreSQL tsvector @@ plainto_tsquery, with ILIKE fallback
+    if search:
+        pattern = f"%{search}%"
+        base = base.where(
+            or_(
+                Company.search_tsv.op("@@")(
+                    func.plainto_tsquery("english", search)
+                ),
+                Company.name.ilike(pattern),
+                Company.industry.ilike(pattern),
+                Company.description.ilike(pattern),
+            )
+        )
+
+    # Sorting
+    sort_col = getattr(Company, sort_by, Company.name)
+    if sort_order == "desc":
+        base = base.order_by(desc(sort_col))
+    else:
+        base = base.order_by(asc(sort_col))
+
+    # Count total (without pagination)
+    count_q = select(func.count()).select_from(base.subquery())
+    total_result = await db.execute(count_q)
+    total = total_result.scalar_one()
+
+    # Paginate
+    offset = (page - 1) * page_size
+    paginated = base.offset(offset).limit(page_size)
+    result = await db.execute(paginated)
+    companies = result.scalars().all()
+
+    return {
+        "items": [_company_to_dict(c) for c in companies],
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+    }
+
+
+async def get_company_detail(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    company_id: uuid.UUID,
+) -> dict[str, Any] | None:
+    """Get a single company with its contacts array."""
+    q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    company = result.scalar_one_or_none()
+    if company is None:
+        return None
+
+    data = _company_to_dict(company, include_contacts=True)
+    # Fetch linked contacts
+    contacts_q = (
+        select(Contact, CompanyContact)
+        .join(CompanyContact, CompanyContact.contact_id == Contact.id)
+        .where(
+            CompanyContact.company_id == company_id,
+            CompanyContact.tenant_id == tenant_id,
+            Contact.deleted_at.is_(None),
+        )
+    )
+    contacts_result = await db.execute(contacts_q)
+    contacts_list = []
+    for contact, link in contacts_result.all():
+        contacts_list.append({
+            "id": str(contact.id),
+            "first_name": contact.first_name,
+            "last_name": contact.last_name,
+            "email": contact.email,
+            "phone": contact.phone,
+            "position": contact.position,
+            "role_at_company": link.role_at_company,
+            "is_primary": link.is_primary,
+        })
+    data["contacts"] = contacts_list
+    return data
+
+
+async def create_company(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any]:
+    """Create a new company and audit-log it."""
+    company = Company(
+        tenant_id=tenant_id,
+        name=data["name"],
+        account_number=data.get("account_number"),
+        industry=data.get("industry"),
+        phone=data.get("phone"),
+        email=data.get("email"),
+        website=data.get("website"),
+        description=data.get("description"),
+        created_by=user_id,
+        updated_by=user_id,
+    )
+    db.add(company)
+    await db.flush()
+    await db.refresh(company)
+    await log_audit(
+        db, tenant_id, user_id, "create", "company", company.id,
+        changes={"name": data["name"]},
+    )
+    return _company_to_dict(company)
+
+
+async def update_company(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    company_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any] | None:
+    """Update a company (partial update) and audit-log changes."""
+    q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    company = result.scalar_one_or_none()
+    if company is None:
+        return None
+
+    changes: dict[str, Any] = {}
+    for field in ("name", "account_number", "industry", "phone", "email", "website", "description"):
+        if field in data and data[field] is not None:
+            old_val = getattr(company, field)
+            changes[field] = {"old": old_val, "new": data[field]}
+            setattr(company, field, data[field])
+    company.updated_by = user_id
+
+    await db.flush()
+    await db.refresh(company)
+    await log_audit(db, tenant_id, user_id, "update", "company", company_id, changes=changes)
+    return _company_to_dict(company)
+
+
+async def soft_delete_company(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    company_id: uuid.UUID,
+    cascade: bool = False,
+) -> bool:
+    """Soft-delete a company. If cascade=True, also soft-delete linked CompanyContact rows."""
+    q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    company = result.scalar_one_or_none()
+    if company is None:
+        return False
+
+    company.deleted_at = datetime.now(timezone.utc)
+    company.updated_by = user_id
+
+    if cascade:
+        # Delete N:M links for this company
+        await db.execute(
+            delete(CompanyContact).where(
+                CompanyContact.company_id == company_id,
+                CompanyContact.tenant_id == tenant_id,
+            )
+        )
+
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "delete", "company", company_id,
+        changes={"name": company.name, "cascade": cascade},
+    )
+    return True
+
+
+async def link_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    company_id: uuid.UUID,
+    contact_id: uuid.UUID,
+    role_at_company: str | None = None,
+    is_primary: bool = False,
+) -> dict[str, Any] | None:
+    """Link a contact to a company (N:M). Returns link data or None if either not found."""
+    # Verify company exists and is in tenant
+    comp_q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    comp_result = await db.execute(comp_q)
+    if comp_result.scalar_one_or_none() is None:
+        return None
+
+    # Verify contact exists and is in tenant
+    cont_q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+    cont_result = await db.execute(cont_q)
+    if cont_result.scalar_one_or_none() is None:
+        return None
+
+    # Check if link already exists
+    existing_q = select(CompanyContact).where(
+        CompanyContact.company_id == company_id,
+        CompanyContact.contact_id == contact_id,
+        CompanyContact.tenant_id == tenant_id,
+    )
+    existing_result = await db.execute(existing_q)
+    existing = existing_result.scalar_one_or_none()
+    if existing is not None:
+        # Update existing link
+        existing.role_at_company = role_at_company
+        existing.is_primary = is_primary
+        await db.flush()
+        await log_audit(
+            db, tenant_id, user_id, "link", "company_contact", existing.id,
+            changes={"company_id": str(company_id), "contact_id": str(contact_id)},
+        )
+        return {
+            "id": str(existing.id),
+            "company_id": str(company_id),
+            "contact_id": str(contact_id),
+            "role_at_company": role_at_company,
+            "is_primary": is_primary,
+        }
+
+    link = CompanyContact(
+        tenant_id=tenant_id,
+        company_id=company_id,
+        contact_id=contact_id,
+        role_at_company=role_at_company,
+        is_primary=is_primary,
+    )
+    db.add(link)
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "link", "company_contact", link.id,
+        changes={"company_id": str(company_id), "contact_id": str(contact_id)},
+    )
+    return {
+        "id": str(link.id),
+        "company_id": str(company_id),
+        "contact_id": str(contact_id),
+        "role_at_company": role_at_company,
+        "is_primary": is_primary,
+    }
+
+
+async def unlink_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    company_id: uuid.UUID,
+    contact_id: uuid.UUID,
+) -> bool:
+    """Unlink a contact from a company (N:M)."""
+    q = select(CompanyContact).where(
+        CompanyContact.company_id == company_id,
+        CompanyContact.contact_id == contact_id,
+        CompanyContact.tenant_id == tenant_id,
+    )
+    result = await db.execute(q)
+    link = result.scalar_one_or_none()
+    if link is None:
+        return False
+
+    await db.delete(link)
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "unlink", "company_contact", link.id,
+        changes={"company_id": str(company_id), "contact_id": str(contact_id)},
+    )
+    return True
+
+
+async def export_companies_csv(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    industry: str | None = None,
+    search: str | None = None,
+) -> str:
+    """Export companies as CSV string."""
+    base = select(Company).where(
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    if industry:
+        base = base.where(Company.industry == industry)
+    if search:
+        pattern = f"%{search}%"
+        base = base.where(
+            or_(
+                Company.search_tsv.op("@@")(
+                    func.plainto_tsquery("english", search)
+                ),
+                Company.name.ilike(pattern),
+                Company.industry.ilike(pattern),
+                Company.description.ilike(pattern),
+            )
+        )
+    base = base.order_by(Company.name)
+    result = await db.execute(base)
+    companies = result.scalars().all()
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(["id", "name", "account_number", "industry", "phone", "email", "website", "description"])
+    for c in companies:
+        writer.writerow([
+            str(c.id), c.name, c.account_number or "", c.industry or "",
+            c.phone or "", c.email or "", c.website or "", c.description or "",
+        ])
+    return output.getvalue()
+
+
+async def export_companies_xlsx(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    industry: str | None = None,
+    search: str | None = None,
+) -> bytes:
+    """Export companies as XLSX bytes."""
+    from openpyxl import Workbook
+
+    base = select(Company).where(
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    if industry:
+        base = base.where(Company.industry == industry)
+    if search:
+        pattern = f"%{search}%"
+        base = base.where(
+            or_(
+                Company.search_tsv.op("@@")(
+                    func.plainto_tsquery("english", search)
+                ),
+                Company.name.ilike(pattern),
+                Company.industry.ilike(pattern),
+                Company.description.ilike(pattern),
+            )
+        )
+    base = base.order_by(Company.name)
+    result = await db.execute(base)
+    companies = result.scalars().all()
+
+    wb = Workbook()
+    ws = wb.active
+    ws.title = "Companies"
+    headers = ["id", "name", "account_number", "industry", "phone", "email", "website", "description"]
+    ws.append(headers)
+    for c in companies:
+        ws.append([
+            str(c.id), c.name, c.account_number or "", c.industry or "",
+            c.phone or "", c.email or "", c.website or "", c.description or "",
+        ])
+
+    buf = io.BytesIO()
+    wb.save(buf)
+    return buf.getvalue()
+
+
+async def get_company_emails(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    company_id: uuid.UUID,
+) -> list[dict[str, Any]]:
+    """Get emails for a company (mail plugin inactive — returns empty array)."""
+    # Verify company exists
+    q = select(Company).where(
+        Company.id == company_id,
+        Company.tenant_id == tenant_id,
+        Company.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    if result.scalar_one_or_none() is None:
+        return []  # Caller should check existence separately
+    return []

app/services/contact_service.py

@@ -0,0 +1,281 @@
+"""Contact service — CRUD, N:M linking, soft-delete, GDPR hard-delete."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import select, func, desc, asc, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.audit import log_audit, log_deletion
+from app.models.contact import Contact, CompanyContact
+from app.models.company import Company
+
+
+def _contact_to_dict(c: Contact, include_companies: bool = False) -> dict[str, Any]:
+    """Serialize a Contact ORM object to dict."""
+    data: dict[str, Any] = {
+        "id": str(c.id),
+        "first_name": c.first_name,
+        "last_name": c.last_name,
+        "email": c.email,
+        "phone": c.phone,
+        "mobile": c.mobile,
+        "position": c.position,
+        "department": c.department,
+        "linkedin_url": c.linkedin_url,
+        "notes": c.notes,
+        "created_at": c.created_at.isoformat() if c.created_at else None,
+        "updated_at": c.updated_at.isoformat() if c.updated_at else None,
+    }
+    if include_companies:
+        data["companies"] = []
+    return data
+
+
+async def list_contacts(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    page: int = 1,
+    page_size: int = 20,
+    search: str | None = None,
+    sort_by: str = "last_name",
+    sort_order: str = "asc",
+) -> dict[str, Any]:
+    """List contacts with pagination and optional search."""
+    page = max(1, page)
+    page_size = max(1, min(100, page_size))
+
+    base = select(Contact).where(
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+
+    if search:
+        pattern = f"%{search}%"
+        base = base.where(
+            (Contact.first_name.ilike(pattern))
+            | (Contact.last_name.ilike(pattern))
+            | (Contact.email.ilike(pattern))
+        )
+
+    sort_col = getattr(Contact, sort_by, Contact.last_name)
+    if sort_order == "desc":
+        base = base.order_by(desc(sort_col))
+    else:
+        base = base.order_by(asc(sort_col))
+
+    count_q = select(func.count()).select_from(base.subquery())
+    total_result = await db.execute(count_q)
+    total = total_result.scalar_one()
+
+    offset = (page - 1) * page_size
+    paginated = base.offset(offset).limit(page_size)
+    result = await db.execute(paginated)
+    contacts = result.scalars().all()
+
+    return {
+        "items": [_contact_to_dict(c) for c in contacts],
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+    }
+
+
+async def get_contact_detail(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    contact_id: uuid.UUID,
+) -> dict[str, Any] | None:
+    """Get a single contact with its linked companies array."""
+    q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    contact = result.scalar_one_or_none()
+    if contact is None:
+        return None
+
+    data = _contact_to_dict(contact, include_companies=True)
+    companies_q = (
+        select(Company, CompanyContact)
+        .join(CompanyContact, CompanyContact.company_id == Company.id)
+        .where(
+            CompanyContact.contact_id == contact_id,
+            CompanyContact.tenant_id == tenant_id,
+            Company.deleted_at.is_(None),
+        )
+    )
+    companies_result = await db.execute(companies_q)
+    companies_list = []
+    for company, link in companies_result.all():
+        companies_list.append({
+            "id": str(company.id),
+            "name": company.name,
+            "industry": company.industry,
+            "role_at_company": link.role_at_company,
+            "is_primary": link.is_primary,
+        })
+    data["companies"] = companies_list
+    return data
+
+
+async def create_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any]:
+    """Create a new contact, optionally linking to companies via company_ids."""
+    contact = Contact(
+        tenant_id=tenant_id,
+        first_name=data["first_name"],
+        last_name=data["last_name"],
+        email=data.get("email"),
+        phone=data.get("phone"),
+        mobile=data.get("mobile"),
+        position=data.get("position"),
+        department=data.get("department"),
+        linkedin_url=data.get("linkedin_url"),
+        notes=data.get("notes"),
+        created_by=user_id,
+        updated_by=user_id,
+    )
+    db.add(contact)
+    await db.flush()
+
+    # Link to companies if company_ids provided
+    company_ids = data.get("company_ids")
+    linked_companies = []
+    if company_ids:
+        for cid_str in company_ids:
+            try:
+                cid = uuid.UUID(cid_str) if isinstance(cid_str, str) else cid_str
+            except (ValueError, TypeError):
+                continue
+            # Verify company exists in tenant
+            comp_q = select(Company).where(
+                Company.id == cid,
+                Company.tenant_id == tenant_id,
+                Company.deleted_at.is_(None),
+            )
+            comp_result = await db.execute(comp_q)
+            if comp_result.scalar_one_or_none() is None:
+                continue
+            link = CompanyContact(
+                tenant_id=tenant_id,
+                company_id=cid,
+                contact_id=contact.id,
+            )
+            db.add(link)
+            linked_companies.append(str(cid))
+        await db.flush()
+
+    await db.refresh(contact)
+    await log_audit(
+        db, tenant_id, user_id, "create", "contact", contact.id,
+        changes={"first_name": data["first_name"], "last_name": data["last_name"], "linked_companies": linked_companies},
+    )
+    return _contact_to_dict(contact)
+
+
+async def update_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    contact_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any] | None:
+    """Update a contact (partial update) and audit-log changes."""
+    q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    contact = result.scalar_one_or_none()
+    if contact is None:
+        return None
+
+    changes: dict[str, Any] = {}
+    for field in ("first_name", "last_name", "email", "phone", "mobile", "position", "department", "linkedin_url", "notes"):
+        if field in data and data[field] is not None:
+            old_val = getattr(contact, field)
+            changes[field] = {"old": old_val, "new": data[field]}
+            setattr(contact, field, data[field])
+    contact.updated_by = user_id
+
+    await db.flush()
+    await db.refresh(contact)
+    await log_audit(db, tenant_id, user_id, "update", "contact", contact_id, changes=changes)
+    return _contact_to_dict(contact)
+
+
+async def soft_delete_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    contact_id: uuid.UUID,
+) -> bool:
+    """Soft-delete a contact (set deleted_at)."""
+    q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    contact = result.scalar_one_or_none()
+    if contact is None:
+        return False
+
+    contact.deleted_at = datetime.now(timezone.utc)
+    contact.updated_by = user_id
+
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "delete", "contact", contact_id,
+        changes={"first_name": contact.first_name, "last_name": contact.last_name},
+    )
+    return True
+
+
+async def gdpr_hard_delete_contact(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    contact_id: uuid.UUID,
+) -> bool:
+    """GDPR hard-delete: physical delete + deletion_log entry with snapshot."""
+    q = select(Contact).where(
+        Contact.id == contact_id,
+        Contact.tenant_id == tenant_id,
+    )
+    result = await db.execute(q)
+    contact = result.scalar_one_or_none()
+    if contact is None:
+        return False
+
+    # Capture snapshot before deletion
+    snapshot = _contact_to_dict(contact)
+
+    # Delete N:M links first
+    await db.execute(
+        delete(CompanyContact).where(
+            CompanyContact.contact_id == contact_id,
+            CompanyContact.tenant_id == tenant_id,
+        )
+    )
+
+    # Physical delete
+    await db.delete(contact)
+    await db.flush()
+
+    # Deletion log (immutable snapshot)
+    await log_deletion(
+        db, tenant_id, user_id, "contact", contact_id, snapshot,
+    )
+    return True

app/services/import_export_service.py

@@ -0,0 +1,216 @@
+"""Import/export service — CSV import with dry-run preview, CSV/XLSX export."""
+
+from __future__ import annotations
+
+import csv
+import io
+import uuid
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.audit import log_audit
+from app.models.company import Company
+from app.models.contact import Contact
+from app.services.company_service import _company_to_dict
+from app.services.contact_service import _contact_to_dict
+
+
+# Expected CSV columns for each entity type
+COMPANY_COLUMNS = ["name", "industry", "phone", "email", "website", "description"]
+CONTACT_COLUMNS = ["first_name", "last_name", "email", "phone", "mobile", "position", "department"]
+
+
+def _parse_csv(content: str) -> list[dict[str, str]]:
+    """Parse CSV content into list of dicts."""
+    reader = csv.DictReader(io.StringIO(content))
+    return [dict(row) for row in reader]
+
+
+def _validate_row(row: dict[str, str], required: list[str]) -> list[str]:
+    """Validate a single row. Returns list of error messages (empty if valid)."""
+    errors = []
+    for col in required:
+        val = row.get(col, "").strip()
+        if not val:
+            errors.append(f"Missing required field: {col}")
+    return errors
+
+
+async def import_companies(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    csv_content: str,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """Import companies from CSV. If dry_run=True, no DB changes are made.
+
+    Returns {total, valid, invalid, errors, created (empty in dry_run)}.
+    """
+    rows = _parse_csv(csv_content)
+    total = len(rows)
+    valid_rows = []
+    errors = []
+
+    for idx, row in enumerate(rows, start=1):
+        row_errors = _validate_row(row, ["name"])
+        if row_errors:
+            for e in row_errors:
+                errors.append({"row": idx, "error": e})
+        else:
+            valid_rows.append(row)
+
+    if dry_run:
+        return {
+            "total": total,
+            "valid": len(valid_rows),
+            "invalid": len(errors),
+            "errors": errors,
+            "created": [],
+            "dry_run": True,
+        }
+
+    created = []
+    for row in valid_rows:
+        company = Company(
+            tenant_id=tenant_id,
+            name=row["name"].strip(),
+            industry=row.get("industry", "").strip() or None,
+            phone=row.get("phone", "").strip() or None,
+            email=row.get("email", "").strip() or None,
+            website=row.get("website", "").strip() or None,
+            description=row.get("description", "").strip() or None,
+            created_by=user_id,
+            updated_by=user_id,
+        )
+        db.add(company)
+        await db.flush()
+        await log_audit(
+            db, tenant_id, user_id, "import", "company", company.id,
+            changes={"name": company.name},
+        )
+        created.append(_company_to_dict(company))
+
+    return {
+        "total": total,
+        "valid": len(valid_rows),
+        "invalid": len(errors),
+        "errors": errors,
+        "created": created,
+        "dry_run": False,
+    }
+
+
+async def import_contacts(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    csv_content: str,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """Import contacts from CSV. If dry_run=True, no DB changes are made.
+
+    Returns {total, valid, invalid, errors, created (empty in dry_run)}.
+    """
+    rows = _parse_csv(csv_content)
+    total = len(rows)
+    valid_rows = []
+    errors = []
+
+    for idx, row in enumerate(rows, start=1):
+        row_errors = _validate_row(row, ["first_name", "last_name"])
+        if row_errors:
+            for e in row_errors:
+                errors.append({"row": idx, "error": e})
+        else:
+            valid_rows.append(row)
+
+    if dry_run:
+        return {
+            "total": total,
+            "valid": len(valid_rows),
+            "invalid": len(errors),
+            "errors": errors,
+            "created": [],
+            "dry_run": True,
+        }
+
+    created = []
+    for row in valid_rows:
+        contact = Contact(
+            tenant_id=tenant_id,
+            first_name=row["first_name"].strip(),
+            last_name=row["last_name"].strip(),
+            email=row.get("email", "").strip() or None,
+            phone=row.get("phone", "").strip() or None,
+            mobile=row.get("mobile", "").strip() or None,
+            position=row.get("position", "").strip() or None,
+            department=row.get("department", "").strip() or None,
+            created_by=user_id,
+            updated_by=user_id,
+        )
+        db.add(contact)
+        await db.flush()
+        await log_audit(
+            db, tenant_id, user_id, "import", "contact", contact.id,
+            changes={"first_name": contact.first_name, "last_name": contact.last_name},
+        )
+        created.append(_contact_to_dict(contact))
+
+    return {
+        "total": total,
+        "valid": len(valid_rows),
+        "invalid": len(errors),
+        "errors": errors,
+        "created": created,
+        "dry_run": False,
+    }
+
+
+async def import_csv(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    csv_content: str,
+    entity_type: str,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """Generic CSV import dispatcher based on entity_type ('companies' or 'contacts')."""
+    if entity_type == "companies":
+        return await import_companies(db, tenant_id, user_id, csv_content, dry_run=dry_run)
+    elif entity_type == "contacts":
+        return await import_contacts(db, tenant_id, user_id, csv_content, dry_run=dry_run)
+    else:
+        return {
+            "total": 0,
+            "valid": 0,
+            "invalid": 0,
+            "errors": [{"row": 0, "error": f"Unknown entity_type: {entity_type}"}],
+            "created": [],
+            "dry_run": dry_run,
+        }
+
+
+async def export_contacts_csv(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+) -> str:
+    """Export contacts as CSV string."""
+    q = select(Contact).where(
+        Contact.tenant_id == tenant_id,
+        Contact.deleted_at.is_(None),
+    ).order_by(Contact.last_name, Contact.first_name)
+    result = await db.execute(q)
+    contacts = result.scalars().all()
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(["id", "first_name", "last_name", "email", "phone", "mobile", "position", "department"])
+    for c in contacts:
+        writer.writerow([
+            str(c.id), c.first_name, c.last_name, c.email or "",
+            c.phone or "", c.mobile or "", c.position or "", c.department or "",
+        ])
+    return output.getvalue()