feat(forward): apply DDL/migration to a database — SECURITY REVIEW REQUIRED

backend/app/api/connections.py

@@ -9,10 +9,16 @@
 
 from app.auth import CurrentUser, get_current_user
 from app.db import get_read_session, get_session
-from app.db_introspect import probe_database
+from app.db_introspect import apply_database_sql, probe_database
 from app.models import DbConnection
 from app.permissions import require_project_member
-from app.schemas import ConnectionCreateIn, ConnectionOut, ConnectionTestOut
+from app.schemas import (
+    ApplySqlIn,
+    ApplySqlOut,
+    ConnectionCreateIn,
+    ConnectionOut,
+    ConnectionTestOut,
+)
 from app.security import decrypt_text, encrypt_text
 from app.sanitize import sanitize_for_storage
 
@@ -104,3 +110,54 @@ async def test_connection(
         return ConnectionTestOut(ok=True, server_version=version, error=None)
     except Exception as exc:  # noqa: BLE001 - message is already DSN-redacted
         return ConnectionTestOut(ok=False, server_version=None, error=str(exc))
+
+
+@router.post("/{db_connection_uuid}/apply-sql", response_model=ApplySqlOut)
+async def apply_sql(
+    db_connection_uuid: uuid.UUID,
+    body: ApplySqlIn,
+    user: CurrentUser = Depends(get_current_user),
+    session: AsyncSession = Depends(get_read_session),
+) -> ApplySqlOut:
+    """Forward engineering: apply DDL/SQL to a stored connection's database.
+
+    SECURITY-SENSITIVE (writes to a live database):
+    * Requires the **editor** role on the connection's project.
+    * IDOR-safe: non-members get a uniform 404 (no enumeration); members
+      lacking editor get 403.
+    * The DSN is decrypted only in memory; the connection reuses the
+      introspectors' SSRF guard (validated + pinned IP).
+    * Runs the whole batch in one transaction; ``dry_run`` (default True) rolls
+      back so nothing persists -- callers must opt in to persist. Failure
+      messages are DSN-redacted, and an execution error is ``ok=false`` rather
+      than an error status (a failed apply is a normal result).
+    """
+    project_space_uuid = await session.scalar(
+        select(DbConnection.project_space_uuid).where(
+            DbConnection.db_connection_uuid == db_connection_uuid
+        )
+    )
+    if project_space_uuid is None:
+        raise HTTPException(status_code=404, detail="connection not found")
+    # Membership first (mask non-members to 404), then require editor (403).
+    try:
+        await require_project_member(
+            session, project_space_uuid, user.user_account_uuid
+        )
+    except HTTPException as exc:
+        if exc.status_code == 403:
+            raise HTTPException(status_code=404, detail="connection not found")
+        raise
+    await require_project_member(
+        session, project_space_uuid, user.user_account_uuid, minimum_role="editor"
+    )
+
+    conn = await session.get(DbConnection, db_connection_uuid)
+    if conn is None:
+        raise HTTPException(status_code=404, detail="connection not found")
+    dsn = decrypt_text(conn.dsn_ciphertext, conn.dsn_nonce)
+    try:
+        await apply_database_sql(dsn, body.sql, dry_run=body.dry_run)
+        return ApplySqlOut(ok=True, dry_run=body.dry_run, error=None)
+    except Exception as exc:  # noqa: BLE001 - message is already DSN-redacted
+        return ApplySqlOut(ok=False, dry_run=body.dry_run, error=str(exc))

backend/app/db_introspect.py

@@ -4,7 +4,11 @@
 from urllib.parse import urlparse
 
 from app.dsn_redaction import redact_dsn_error_message
-from app.pg_introspect.introspect import introspect_postgres, probe_postgres
+from app.pg_introspect.introspect import (
+    apply_postgres_sql,
+    introspect_postgres,
+    probe_postgres,
+)
 from app.snowflake_introspect import introspect_snowflake
 from app.snowflake_introspect.introspect import probe_snowflake
 
@@ -50,3 +54,20 @@ async def probe_database(dsn: str) -> str:
     except Exception as exc:
         message = str(exc) or type(exc).__name__
         raise RuntimeError(redact_dsn_error_message(message, dsn)) from None
+
+
+async def apply_database_sql(dsn: str, sql: str, dry_run: bool = True) -> None:
+    """Forward engineering: apply DDL/SQL to a target database.
+
+    PostgreSQL only for now (Snowflake DDL apply differs and is connector-gated).
+    Errors are DSN-redacted so credentials never surface in an API response.
+    """
+
+    try:
+        dialect = detect_dsn_dialect(dsn)
+        if dialect != "postgresql":
+            raise ValueError("forward apply is only supported for PostgreSQL")
+        await apply_postgres_sql(dsn, sql, dry_run=dry_run)
+    except Exception as exc:
+        message = str(exc) or type(exc).__name__
+        raise RuntimeError(redact_dsn_error_message(message, dsn)) from None

backend/app/pg_introspect/introspect.py

@@ -33,6 +33,47 @@ async def probe_postgres(dsn: str) -> str:
         await conn.close()
 
 
+async def apply_postgres_sql(dsn: str, sql: str, dry_run: bool = True) -> None:
+    """Execute DDL/SQL against a target Postgres DB inside one transaction.
+
+    Forward engineering: materialize a designed schema / apply a migration.
+    SSRF-guarded exactly like introspection (``validate_postgres_dsn_target`` +
+    pinned IP). The whole batch runs in a single transaction:
+
+    * ``dry_run=True`` (default) rolls back, so nothing persists -- a pre-flight
+      validation that the SQL executes cleanly against the real database.
+    * ``dry_run=False`` commits.
+
+    PostgreSQL DDL is transactional, so a mid-batch failure rolls everything
+    back. Statements that cannot run inside a transaction (e.g.
+    ``CREATE INDEX CONCURRENTLY``) will raise here -- that is expected.
+    """
+    target = await validate_postgres_dsn_target(dsn)
+    connect_host: str | list[str] = (
+        target.hosts[0] if len(target.hosts) == 1 else list(target.hosts)
+    )
+    if target.port is not None:
+        conn = await asyncpg.connect(
+            dsn, host=connect_host, port=target.port, timeout=15
+        )
+    else:
+        conn = await asyncpg.connect(dsn, host=connect_host, timeout=15)
+    try:
+        tx = conn.transaction()
+        await tx.start()
+        try:
+            await conn.execute(sql)
+        except BaseException:
+            await tx.rollback()
+            raise
+        if dry_run:
+            await tx.rollback()
+        else:
+            await tx.commit()
+    finally:
+        await conn.close()
+
+
 async def introspect_postgres(dsn: str, schema_filter: str | None) -> dict:
     """Introspect a PostgreSQL database and return a snapshot JSON."""
 

backend/app/schemas.py

@@ -67,6 +67,22 @@ class ConnectionTestOut(BaseModel):
     error: str | None = None
 
 
+class ApplySqlIn(BaseModel):
+    """Request body for forward-engineering: apply DDL/SQL to a connection."""
+
+    sql: str = Field(min_length=1, max_length=262_144)
+    # Default to a rolled-back pre-flight; the caller must opt in to persist.
+    dry_run: bool = True
+
+
+class ApplySqlOut(BaseModel):
+    """Result of applying SQL (DSN-redacted on failure)."""
+
+    ok: bool
+    dry_run: bool
+    error: str | None = None
+
+
 class SnapshotCreateIn(BaseModel):
     """Request body for creating a schema snapshot."""
 

backend/tests/test_api_apply_sql.py

@@ -0,0 +1,123 @@
+import uuid
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from app.api.connections import apply_sql
+from app.auth import CurrentUser
+from app.db_introspect import apply_database_sql
+from app.schemas import ApplySqlIn
+
+
+def _user():
+    return CurrentUser(user_account_uuid=uuid.uuid4(), subject="t", display_name="T")
+
+
+def _conn():
+    return SimpleNamespace(dsn_ciphertext=b"c", dsn_nonce=b"n")
+
+
+def _body(dry_run=True):
+    return ApplySqlIn(sql="CREATE TABLE t (id bigint);", dry_run=dry_run)
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_returns_404_when_missing():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=None)
+    with pytest.raises(HTTPException) as e:
+        await apply_sql(db_connection_uuid=uuid.uuid4(), body=_body(), user=_user(), session=session)
+    assert e.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_masks_non_member_as_404():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=uuid.uuid4())
+    with patch(
+        "app.api.connections.require_project_member",
+        new_callable=AsyncMock,
+        side_effect=HTTPException(status_code=403, detail="project access denied"),
+    ):
+        with pytest.raises(HTTPException) as e:
+            await apply_sql(db_connection_uuid=uuid.uuid4(), body=_body(), user=_user(), session=session)
+    assert e.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_returns_403_when_member_lacks_editor():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=uuid.uuid4())
+    # first call (membership) passes, second call (editor) raises 403
+    with patch(
+        "app.api.connections.require_project_member",
+        new_callable=AsyncMock,
+        side_effect=[None, HTTPException(status_code=403, detail="insufficient project role")],
+    ):
+        with pytest.raises(HTTPException) as e:
+            await apply_sql(db_connection_uuid=uuid.uuid4(), body=_body(), user=_user(), session=session)
+    assert e.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_reports_ok_true_on_success():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=uuid.uuid4())
+    session.get = AsyncMock(return_value=_conn())
+    with patch(
+        "app.api.connections.require_project_member", new_callable=AsyncMock
+    ), patch(
+        "app.api.connections.decrypt_text", return_value="postgresql://u@db.example.com/x"
+    ), patch(
+        "app.api.connections.apply_database_sql", new_callable=AsyncMock
+    ) as apply_mock:
+        out = await apply_sql(
+            db_connection_uuid=uuid.uuid4(), body=_body(dry_run=True), user=_user(), session=session
+        )
+    assert out.ok is True and out.dry_run is True and out.error is None
+    apply_mock.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_apply_sql_reports_ok_false_on_error():
+    session = AsyncMock()
+    session.scalar = AsyncMock(return_value=uuid.uuid4())
+    session.get = AsyncMock(return_value=_conn())
+    with patch(
+        "app.api.connections.require_project_member", new_callable=AsyncMock
+    ), patch(
+        "app.api.connections.decrypt_text", return_value="postgresql://u@db.example.com/x"
+    ), patch(
+        "app.api.connections.apply_database_sql",
+        new_callable=AsyncMock,
+        side_effect=RuntimeError("syntax error at or near"),
+    ):
+        out = await apply_sql(
+            db_connection_uuid=uuid.uuid4(), body=_body(dry_run=False), user=_user(), session=session
+        )
+    assert out.ok is False and out.dry_run is False and "syntax error" in (out.error or "")
+
+
+@pytest.mark.asyncio
+async def test_apply_database_sql_rejects_non_postgres():
+    with pytest.raises(RuntimeError):
+        await apply_database_sql("snowflake://u@acct/db", "CREATE TABLE t (id int);")
+
+
+@pytest.mark.asyncio
+async def test_apply_database_sql_routes_postgres_and_redacts():
+    with patch("app.db_introspect.apply_postgres_sql", new_callable=AsyncMock) as m:
+        await apply_database_sql(
+            "postgresql://u@db.example.com/x", "CREATE TABLE t (id int);", dry_run=True
+        )
+    m.assert_awaited_once()
+    with patch(
+        "app.db_introspect.apply_postgres_sql",
+        new_callable=AsyncMock,
+        side_effect=OSError("connect to postgresql://user:s3cret@db"),
+    ):
+        with pytest.raises(RuntimeError) as e:
+            await apply_database_sql("postgresql://user:s3cret@db.example.com/x", "X")
+    assert "s3cret" not in str(e.value)

docs/valuation-gap-analysis.md

@@ -40,6 +40,11 @@ now cross into **act** (migration SQL). The gaps below are ordered by value-per-
 - **Data dictionary export** — Markdown/HTML doc of schema + **annotations** (now that annotations exist) + PK/FK/index. High value for the "living documentation" buyer; pure-logic, testable.
 - **Relationship inference** — suggest implicit FKs by naming/type heuristics (`*_id` → `id`). Composes with existing cardinality logic.
 
+### P0b — Forward engineering (closes the round-trip: model → database)
+The product was reverse-only (DB → model). **Forward engineering** — materializing a designed schema / applying a migration to a real database — is the other half and a major value unlock: design → deploy in one tool.
+- **Apply DDL/migration to a connection** _(backend delivered — SECURITY REVIEW REQUIRED)_: `POST /api/connections/{uuid}/apply-sql` runs DDL against the connection's DB inside one transaction, **dry-run by default** (rolls back = pre-flight validation), editor-only, IDOR-safe, SSRF-guarded (reuses `validate_postgres_dsn_target` + pinned IP), DSN-redacted errors. Because it writes to a live DB, it must pass human security review before merge and needs a careful confirmation UI.
+- **Next**: down-migration generation (up+down = full change management), framework-formatted migrations (Alembic/Flyway), and a guarded "deploy schema" UX with explicit confirm + dry-run preview.
+
 ### P2 — Scale, governance, monetization (justifies a valuation, not just a product)
 - **Robust async jobs** — `jobs` is DB-queue MVP; harden the valkey worker path (retries/backoff/idempotency) for large-schema introspection at scale.
 - **Audit log + org multi-tenancy + usage quotas** — table-stakes for enterprise/SOC2 and for billing.