feat: Schema Migration Service

Dockerfile

@@ -14,6 +14,11 @@ COPY . .
 ARG GIT_COMMIT=unknown
 RUN GIT_COMMIT=${GIT_COMMIT} pnpm build
 
+ARG TARGETARCH=amd64
+RUN unzip bin/pgschema-linux-${TARGETARCH}.zip -d /usr/local/bin \
+  && mv /usr/local/bin/pgschema-linux-${TARGETARCH} /usr/local/bin/pgschema \
+  && chmod +x /usr/local/bin/pgschema
+
 # Runtime dependencies
 # Generated from esbuild externals + package.json versions (single source of truth).
 # Only rebuilds when package.json or build-server.mjs externals change.
@@ -43,13 +48,16 @@ RUN node scripts/gen-runtime-package.mjs > runtime-package.json \
 # Layers ordered least → most frequently changing for cache efficiency
 FROM alpine:3.21
 
-RUN apk add --no-cache libstdc++
+RUN apk add --no-cache libstdc++ git
 
 COPY --from=node:22-alpine /usr/local/bin/node /usr/local/bin/node
 
+RUN addgroup -S pgconsole && adduser -S pgconsole -G pgconsole
+
 WORKDIR /app
 
-# 1. Entrypoint — rarely changes
+# 1. Binaries & entrypoint — rarely change
+COPY --from=builder /usr/local/bin/pgschema /usr/local/bin/pgschema
 COPY docker-entrypoint.sh /app/
 
 # 2. Runtime node_modules — changes only when externals or dep versions change
@@ -74,4 +82,6 @@ ENV NODE_ENV=production
 ENV PORT=9876
 EXPOSE 9876
 
+USER pgconsole
+
 ENTRYPOINT ["/app/docker-entrypoint.sh"]

pgconsole.example.toml

@@ -96,6 +96,12 @@ username = "app_user"
 password = "staging_password"
 ssl_mode = "require"
 labels = ["staging"]
+# Schema migration source (optional) — enables migration features
+# [connections.schema_source]
+# repo = "https://github.com/myorg/db-schema.git"
+# branch = "main"
+# path = "schema/main.sql"
+# schema = "public"
 
 # Example with full SSL configuration and timeouts
 # [[connections]]

proto/migration.proto

@@ -0,0 +1,56 @@
+syntax = "proto3";
+
+package migration.v1;
+
+service MigrationService {
+  rpc PlanMigration(PlanMigrationRequest) returns (PlanMigrationResponse);
+  rpc ApplyMigration(ApplyMigrationRequest) returns (stream ApplyMigrationResponse);
+  rpc GetSchemaSourceStatus(GetSchemaSourceStatusRequest) returns (GetSchemaSourceStatusResponse);
+}
+
+message PlanMigrationRequest {
+  string connection_id = 1;
+}
+
+message SchemaDiff {
+  string sql = 1;
+  string type = 2;
+  string operation = 3;
+  string path = 4;
+  bool can_run_in_transaction = 5;
+}
+
+message PlanMigrationResponse {
+  string plan_id = 1;
+  string branch = 2;
+  string commit_hash = 3;
+  string source_fingerprint = 4;
+  repeated SchemaDiff diffs = 5;
+  bool can_run_in_transaction = 6;
+  string summary = 7;
+}
+
+message ApplyMigrationRequest {
+  string connection_id = 1;
+  string plan_id = 2;
+}
+
+message ApplyMigrationResponse {
+  int32 step = 1;
+  int32 total_steps = 2;
+  string sql = 3;
+  string status = 4;
+  string error = 5;
+}
+
+message GetSchemaSourceStatusRequest {
+  string connection_id = 1;
+}
+
+message GetSchemaSourceStatusResponse {
+  bool configured = 1;
+  string repo = 2;
+  string branch = 3;
+  string path = 4;
+  string schema = 5;
+}

server/connect.ts

@@ -1,14 +1,26 @@
 import { expressConnectMiddleware } from "@connectrpc/connect-express";
+import type { Interceptor } from "@connectrpc/connect";
 import type { Request } from "express";
 import { ConnectionService } from "../src/gen/connection_connect";
 import { QueryService } from "../src/gen/query_connect";
 import { AIService } from "../src/gen/ai_connect";
+import { MigrationService } from "../src/gen/migration_connect";
 import { connectionServiceHandlers } from "./services/connection-service";
 import { queryServiceHandlers } from "./services/query-service";
 import { aiServiceHandlers } from "./services/ai-service";
+import { migrationServiceHandlers } from "./services/migration-service";
 import { getCurrentUser, type User } from "./lib/auth";
 import { isAuthEnabled } from "./lib/config";
 
+const loggingInterceptor: Interceptor = (next) => async (req) => {
+  try {
+    return await next(req)
+  } catch (err) {
+    console.error(`[RPC] ${req.service.typeName}/${req.method.name}:`, err)
+    throw err
+  }
+}
+
 // Helper to get user from ConnectRPC context
 // Note: contextValues may be a Promise if contextValues factory is async
 export async function getUserFromContext(contextValues: Map<string, unknown> | Promise<Map<string, unknown>>): Promise<User | null> {
@@ -25,9 +37,10 @@ const GUEST_USER: User = { email: 'guest', name: 'Guest' }
  */
 export const connectRouter = expressConnectMiddleware({
   routes: (router) => {
-    router.service(ConnectionService, connectionServiceHandlers);
-    router.service(QueryService, queryServiceHandlers);
-    router.service(AIService, aiServiceHandlers);
+    router.service(ConnectionService, connectionServiceHandlers, { interceptors: [loggingInterceptor] });
+    router.service(QueryService, queryServiceHandlers, { interceptors: [loggingInterceptor] });
+    router.service(AIService, aiServiceHandlers, { interceptors: [loggingInterceptor] });
+    router.service(MigrationService, migrationServiceHandlers, { interceptors: [loggingInterceptor] });
   },
   // Set max message size to ~4GB for large query results
   readMaxBytes: 0xffffffff,

server/index.ts

@@ -68,7 +68,8 @@ app.use((req, res, next) => {
       req.path.startsWith('/api/') ||
       req.path.startsWith('/connection.v1.') ||
       req.path.startsWith('/query.v1.') ||
-      req.path.startsWith('/ai.v1.')) {
+      req.path.startsWith('/ai.v1.') ||
+      req.path.startsWith('/migration.v1.')) {
     return next()
   }
   res.sendFile(path.join(clientDir, 'index.html'))

server/lib/config.ts

@@ -10,6 +10,13 @@ export interface LabelConfig {
   color: string
 }
 
+export interface SchemaSourceConfig {
+  repo: string
+  branch?: string
+  path: string
+  schema: string
+}
+
 export interface ConnectionConfig {
   id: string
   name: string
@@ -26,6 +33,7 @@ export interface ConnectionConfig {
   lock_timeout?: string
   statement_timeout?: string
   lazy?: boolean
+  schema_source?: SchemaSourceConfig
 }
 
 export interface UserConfig {
@@ -406,6 +414,24 @@ export async function loadConfigFromString(content: string): Promise<void> {
       }
     }
 
+    // Parse schema_source if provided
+    let schemaSource: SchemaSourceConfig | undefined = undefined
+    const rawSchemaSource = c.schema_source as Record<string, unknown> | undefined
+    if (rawSchemaSource) {
+      if (!rawSchemaSource.repo || typeof rawSchemaSource.repo !== 'string') {
+        throw new Error(`Connection ${c.id} schema_source.repo is required and must be a string`)
+      }
+      if (!rawSchemaSource.path || typeof rawSchemaSource.path !== 'string') {
+        throw new Error(`Connection ${c.id} schema_source.path is required and must be a string`)
+      }
+      schemaSource = {
+        repo: rawSchemaSource.repo,
+        branch: typeof rawSchemaSource.branch === 'string' ? rawSchemaSource.branch : undefined,
+        path: rawSchemaSource.path,
+        schema: typeof rawSchemaSource.schema === 'string' ? rawSchemaSource.schema : 'public',
+      }
+    }
+
     connections.push({
       id: c.id,
       name: c.name,
@@ -422,6 +448,7 @@ export async function loadConfigFromString(content: string): Promise<void> {
       lock_timeout: typeof c.lock_timeout === 'string' ? c.lock_timeout : undefined,
       statement_timeout: typeof c.statement_timeout === 'string' ? c.statement_timeout : undefined,
       lazy: c.lazy === true,
+      schema_source: schemaSource,
     })
   }
 

server/lib/git.ts

@@ -0,0 +1,70 @@
+import { execFile } from 'child_process'
+import { access, rm } from 'fs/promises'
+import { join } from 'path'
+import { tmpdir } from 'os'
+
+function exec(cmd: string, args: string[], cwd?: string): Promise<string> {
+  return new Promise((resolve, reject) => {
+    execFile(cmd, args, { cwd, timeout: 60_000 }, (error, stdout, stderr) => {
+      if (error) {
+        reject(new Error(`git ${args[0]} failed: ${stderr || error.message}`))
+      } else {
+        resolve(stdout.trim())
+      }
+    })
+  })
+}
+
+// Cache the repo URL per directory so we can detect config changes
+const repoDirUrls = new Map<string, string>()
+// Per-connection lock to prevent concurrent clone/fetch races
+const syncLocks = new Map<string, Promise<{ commitHash: string }>>()
+
+export function getRepoDir(connectionId: string): string {
+  return join(tmpdir(), 'pgconsole-schema', connectionId)
+}
+
+export async function syncRepo(connectionId: string, repo: string, branch?: string): Promise<{ commitHash: string }> {
+  // Serialize concurrent sync requests for the same connection
+  const existing = syncLocks.get(connectionId)
+  if (existing) {
+    return existing
+  }
+
+  const promise = doSyncRepo(connectionId, repo, branch).finally(() => {
+    syncLocks.delete(connectionId)
+  })
+  syncLocks.set(connectionId, promise)
+  return promise
+}
+
+async function doSyncRepo(connectionId: string, repo: string, branch?: string): Promise<{ commitHash: string }> {
+  const repoDir = getRepoDir(connectionId)
+
+  const exists = await access(join(repoDir, '.git')).then(() => true).catch(() => false)
+
+  // If the repo URL changed, wipe the old checkout
+  if (exists) {
+    const cachedUrl = repoDirUrls.get(repoDir)
+    if (cachedUrl && cachedUrl !== repo) {
+      await rm(repoDir, { recursive: true, force: true })
+    }
+  }
+
+  const stillExists = await access(join(repoDir, '.git')).then(() => true).catch(() => false)
+
+  if (stillExists) {
+    await exec('git', ['fetch', 'origin', ...(branch ? [branch] : [])], repoDir)
+    await exec('git', ['reset', '--hard', branch ? `origin/${branch}` : 'FETCH_HEAD'], repoDir)
+  } else {
+    const cloneArgs = ['clone', '--depth', '1']
+    if (branch) cloneArgs.push('--branch', branch)
+    cloneArgs.push(repo, repoDir)
+    await exec('git', cloneArgs)
+  }
+
+  repoDirUrls.set(repoDir, repo)
+
+  const commitHash = await exec('git', ['rev-parse', 'HEAD'], repoDir)
+  return { commitHash }
+}