feat: phase 2 databases via DBX (v0.3.0)

- dbx-mcp persistent stdio subprocess in Go backend (internal/dbx),
  lazy start + single retry, tool failures surface as 502
- core.db_connections registry: per-project connections, credentials
  server-side only, managed entries protected (409)
- platform-postgres + platform-dragonfly auto-register per project
- REST proxy: list/create/delete/test, databases, tables, describe,
  schema context, SQL query, redis command
- Databases page: connection list, schema/table browser, column
  inspector, SQL editor with result grid + row detail, redis console
- backend image moved to debian-slim for glibc; pinned dbx release
  with SHA256 verification

Authored By: TDvorak <[email protected]>
This commit is contained in:
Tomas Dvorak committed 2026-09-16 14:54:33 +02:00
1 parent 420502eec1
commit 627bd89d5e
71 files changed
+3685 -49

No files matched your search

+42 -1
View File
@@ -4,7 +4,10 @@ import (
"context"
"fmt"
"log/slog"
"net/url"
"os"
"strconv"
"strings"
"time"
"github.com/gin-gonic/gin"
@@ -16,6 +19,7 @@ import (
"github.com/tdvorak/primora/apps/backend/internal/auth"
"github.com/tdvorak/primora/apps/backend/internal/config"
"github.com/tdvorak/primora/apps/backend/internal/database"
"github.com/tdvorak/primora/apps/backend/internal/dbx"
"github.com/tdvorak/primora/apps/backend/internal/handlers"
"github.com/tdvorak/primora/apps/backend/internal/middleware"
"github.com/tdvorak/primora/apps/backend/internal/observability"
@@ -30,6 +34,7 @@ type App struct {
Logger *slog.Logger
DB *pgxpool.Pool
Redis *redis.Client
DBX *dbx.Client
}
func Bootstrap(ctx context.Context) (*App, error) {
@@ -75,8 +80,9 @@ func Bootstrap(ctx context.Context) (*App, error) {
if err != nil {
return nil, err
}
dbxClient := dbx.NewClient(logger, cfg.DBXDataDir)
repo := repositories.NewCoreRepository(dbPool)
platform := services.NewPlatformService(repo, store, services.NewMailer(cfg), os.Getenv("VITE_APP_URL"))
platform := services.NewPlatformService(repo, store, services.NewMailer(cfg), os.Getenv("VITE_APP_URL"), dbxClient, firstPartyDBSeeds(cfg))
if cfg.Env == "production" {
gin.SetMode(gin.ReleaseMode)
@@ -148,6 +154,7 @@ func Bootstrap(ctx context.Context) (*App, error) {
Logger: logger,
DB: dbPool,
Redis: redisClient,
DBX: dbxClient,
}, nil
}
@@ -157,7 +164,41 @@ func (a *App) Run() error {
return a.Router.Run(address)
}
// firstPartyDBSeeds turns the platform's own Postgres and Dragonfly URLs into
// managed connection seeds registered on every project.
func firstPartyDBSeeds(cfg config.Config) []services.ManagedDBSeed {
var seeds []services.ManagedDBSeed
if u, err := url.Parse(cfg.DatabaseURL); err == nil && u.Hostname() != "" {
cfg2 := services.DBConnectionConfig{
Host: u.Hostname(),
Database: strings.TrimPrefix(u.Path, "/"),
Username: u.User.Username(),
}
if p, ok := u.User.Password(); ok {
cfg2.Password = p
}
if port, err := strconv.Atoi(u.Port()); err == nil {
cfg2.Port = &port
}
seeds = append(seeds, services.ManagedDBSeed{Name: "platform-postgres", DBType: "postgres", Config: cfg2})
}
if u, err := url.Parse(cfg.DragonflyURL); err == nil && u.Hostname() != "" {
cfg2 := services.DBConnectionConfig{Host: u.Hostname()}
if p, ok := u.User.Password(); ok {
cfg2.Password = p
}
if port, err := strconv.Atoi(u.Port()); err == nil {
cfg2.Port = &port
}
seeds = append(seeds, services.ManagedDBSeed{Name: "platform-dragonfly", DBType: "redis", Config: cfg2})
}
return seeds
}
func (a *App) Close() error {
if a.DBX != nil {
a.DBX.Close()
}
if a.Redis != nil {
if err := a.Redis.Close(); err != nil {
return fmt.Errorf("close redis: %w", err)
+7
View File
@@ -4,6 +4,7 @@ import (
"errors"
"fmt"
"os"
"path/filepath"
"strconv"
"strings"
)
@@ -14,6 +15,7 @@ type Config struct {
DatabaseURL string
DragonflyURL string
StorageRoot string
DBXDataDir string
AuthInternalBaseURL string
PublicURL string
UserRateLimitPerMin int
@@ -40,6 +42,7 @@ func Load() (Config, error) {
DatabaseURL: os.Getenv("DATABASE_URL"),
DragonflyURL: getenv("DRAGONFLY_URL", "redis://localhost:6379/0"),
StorageRoot: getenv("BACKEND_STORAGE_ROOT", "./tmp/storage"),
DBXDataDir: os.Getenv("DBX_DATA_DIR"),
AuthInternalBaseURL: getenv("AUTH_INTERNAL_BASE_URL", "http://auth:3001"),
PublicURL: getenv("VITE_APP_URL", "http://localhost"),
UserRateLimitPerMin: 240,
@@ -104,6 +107,10 @@ func Load() (Config, error) {
return Config{}, errors.New("missing required environment values: " + strings.Join(missing, ", "))
}
if cfg.DBXDataDir == "" {
cfg.DBXDataDir = filepath.Join(cfg.StorageRoot, "dbx")
}
return cfg, nil
}
@@ -0,0 +1,173 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.29.0
// source: db_connections.sql
package db
import (
"context"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
)
const createDBConnection = `-- name: CreateDBConnection :one
INSERT INTO core.db_connections (
project_id,
name,
db_type,
config,
is_managed,
created_by_user_id
) VALUES ($1, $2, $3, $4, $5, $6)
RETURNING id, project_id, name, db_type, config, is_managed, created_by_user_id, created_at, updated_at
`
type CreateDBConnectionParams struct {
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
DbType string `json:"db_type"`
Config []byte `json:"config"`
IsManaged bool `json:"is_managed"`
CreatedByUserID pgtype.UUID `json:"created_by_user_id"`
}
func (q *Queries) CreateDBConnection(ctx context.Context, arg CreateDBConnectionParams) (CoreDbConnection, error) {
row := q.db.QueryRow(ctx, createDBConnection,
arg.ProjectID,
arg.Name,
arg.DbType,
arg.Config,
arg.IsManaged,
arg.CreatedByUserID,
)
var i CoreDbConnection
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.DbType,
&i.Config,
&i.IsManaged,
&i.CreatedByUserID,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const deleteDBConnection = `-- name: DeleteDBConnection :one
DELETE FROM core.db_connections
WHERE id = $1
AND project_id = $2
RETURNING id, project_id, name, db_type, config, is_managed, created_by_user_id, created_at, updated_at
`
type DeleteDBConnectionParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
}
func (q *Queries) DeleteDBConnection(ctx context.Context, arg DeleteDBConnectionParams) (CoreDbConnection, error) {
row := q.db.QueryRow(ctx, deleteDBConnection, arg.ID, arg.ProjectID)
var i CoreDbConnection
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.DbType,
&i.Config,
&i.IsManaged,
&i.CreatedByUserID,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const getDBConnectionByID = `-- name: GetDBConnectionByID :one
SELECT id, project_id, name, db_type, config, is_managed, created_by_user_id, created_at, updated_at FROM core.db_connections
WHERE id = $1
`
func (q *Queries) GetDBConnectionByID(ctx context.Context, id uuid.UUID) (CoreDbConnection, error) {
row := q.db.QueryRow(ctx, getDBConnectionByID, id)
var i CoreDbConnection
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.DbType,
&i.Config,
&i.IsManaged,
&i.CreatedByUserID,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const getDBConnectionByName = `-- name: GetDBConnectionByName :one
SELECT id, project_id, name, db_type, config, is_managed, created_by_user_id, created_at, updated_at FROM core.db_connections
WHERE project_id = $1
AND name = $2
`
type GetDBConnectionByNameParams struct {
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
}
func (q *Queries) GetDBConnectionByName(ctx context.Context, arg GetDBConnectionByNameParams) (CoreDbConnection, error) {
row := q.db.QueryRow(ctx, getDBConnectionByName, arg.ProjectID, arg.Name)
var i CoreDbConnection
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.DbType,
&i.Config,
&i.IsManaged,
&i.CreatedByUserID,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const listDBConnections = `-- name: ListDBConnections :many
SELECT id, project_id, name, db_type, config, is_managed, created_by_user_id, created_at, updated_at FROM core.db_connections
WHERE project_id = $1
ORDER BY is_managed DESC, name ASC
`
func (q *Queries) ListDBConnections(ctx context.Context, projectID uuid.UUID) ([]CoreDbConnection, error) {
rows, err := q.db.Query(ctx, listDBConnections, projectID)
if err != nil {
return nil, err
}
defer rows.Close()
var items []CoreDbConnection
for rows.Next() {
var i CoreDbConnection
if err := rows.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.DbType,
&i.Config,
&i.IsManaged,
&i.CreatedByUserID,
&i.CreatedAt,
&i.UpdatedAt,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
@@ -200,6 +200,18 @@ type CoreCollection struct {
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
}
type CoreDbConnection struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
DbType string `json:"db_type"`
Config []byte `json:"config"`
IsManaged bool `json:"is_managed"`
CreatedByUserID pgtype.UUID `json:"created_by_user_id"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
}
type CoreDocument struct {
ID uuid.UUID `json:"id"`
CollectionID uuid.UUID `json:"collection_id"`
@@ -25,6 +25,7 @@ type Querier interface {
CreateBucket(ctx context.Context, arg CreateBucketParams) (CoreBucket, error)
CreateBucketObject(ctx context.Context, arg CreateBucketObjectParams) (CoreBucketObject, error)
CreateCollection(ctx context.Context, arg CreateCollectionParams) (CoreCollection, error)
CreateDBConnection(ctx context.Context, arg CreateDBConnectionParams) (CoreDbConnection, error)
CreateDocument(ctx context.Context, arg CreateDocumentParams) (CoreDocument, error)
CreateInvitation(ctx context.Context, arg CreateInvitationParams) (CoreProjectInvitation, error)
CreateOrganization(ctx context.Context, arg CreateOrganizationParams) (CoreOrganization, error)
@@ -32,6 +33,7 @@ type Querier interface {
DeleteBucketByID(ctx context.Context, id uuid.UUID) (CoreBucket, error)
DeleteBucketObjectByKey(ctx context.Context, arg DeleteBucketObjectByKeyParams) (CoreBucketObject, error)
DeleteCollection(ctx context.Context, arg DeleteCollectionParams) error
DeleteDBConnection(ctx context.Context, arg DeleteDBConnectionParams) (CoreDbConnection, error)
DeleteDocument(ctx context.Context, arg DeleteDocumentParams) error
DeleteOrganizationByID(ctx context.Context, id uuid.UUID) (CoreOrganization, error)
DeletePendingInvitationByIDForOrganization(ctx context.Context, arg DeletePendingInvitationByIDForOrganizationParams) (CoreProjectInvitation, error)
@@ -42,6 +44,8 @@ type Querier interface {
GetBucketObjectByKey(ctx context.Context, arg GetBucketObjectByKeyParams) (CoreBucketObject, error)
GetCollectionByID(ctx context.Context, id uuid.UUID) (CoreCollection, error)
GetCollectionBySlug(ctx context.Context, arg GetCollectionBySlugParams) (CoreCollection, error)
GetDBConnectionByID(ctx context.Context, id uuid.UUID) (CoreDbConnection, error)
GetDBConnectionByName(ctx context.Context, arg GetDBConnectionByNameParams) (CoreDbConnection, error)
GetDocumentByID(ctx context.Context, arg GetDocumentByIDParams) (CoreDocument, error)
GetInvitationByIDForOrganization(ctx context.Context, arg GetInvitationByIDForOrganizationParams) (CoreProjectInvitation, error)
GetInvitationByTokenHash(ctx context.Context, tokenHash string) (CoreProjectInvitation, error)
@@ -57,6 +61,7 @@ type Querier interface {
ListBucketsForOrganization(ctx context.Context, organizationID uuid.UUID) ([]uuid.UUID, error)
ListBucketsForProject(ctx context.Context, arg ListBucketsForProjectParams) ([]CoreBucket, error)
ListCollections(ctx context.Context, projectID uuid.UUID) ([]CoreCollection, error)
ListDBConnections(ctx context.Context, projectID uuid.UUID) ([]CoreDbConnection, error)
ListDocuments(ctx context.Context, arg ListDocumentsParams) ([]CoreDocument, error)
ListInvitationsForOrganization(ctx context.Context, organizationID uuid.UUID) ([]ListInvitationsForOrganizationRow, error)
ListOrganizationMembers(ctx context.Context, organizationID uuid.UUID) ([]ListOrganizationMembersRow, error)
+318
View File
@@ -0,0 +1,318 @@
// Package dbx bridges Primora to the DBX MCP server (@dbx-app/mcp-server).
// The server is spawned lazily as a persistent stdio subprocess; all database
// inspection and queries flow through its tools.
package dbx
import (
"bytes"
"context"
"fmt"
"log/slog"
"os"
"os/exec"
"strings"
"sync"
"time"
"github.com/modelcontextprotocol/go-sdk/mcp"
)
type Client struct {
mu sync.Mutex
session *mcp.ClientSession
cmd *exec.Cmd
cmdLine string
dataDir string
logger *slog.Logger
registered map[string]bool
lastErr string
}
func NewClient(logger *slog.Logger, dataDir string) *Client {
return &Client{logger: logger, dataDir: dataDir, registered: map[string]bool{}}
}
func (d *Client) connect(ctx context.Context) (*mcp.ClientSession, error) {
d.mu.Lock()
defer d.mu.Unlock()
if d.session != nil {
return d.session, nil
}
// jarvis: exec.Command, not CommandContext — the subprocess must outlive the
// request ctx or every HTTP request would SIGKILL it on return.
var cmd *exec.Cmd
if p, err := exec.LookPath("dbx-mcp"); err == nil {
cmd = exec.Command(p)
d.cmdLine = "dbx-mcp"
} else if p, err := exec.LookPath("dbx-mcp-server"); err == nil {
cmd = exec.Command(p)
d.cmdLine = "dbx-mcp-server"
} else {
npx, err := exec.LookPath("npx")
if err != nil {
d.lastErr = "neither dbx-mcp binary nor npx found — install @dbx-app/mcp-server"
return nil, fmt.Errorf("%s", d.lastErr)
}
cmd = exec.Command(npx, "-y", "@dbx-app/mcp-server")
d.cmdLine = "npx -y @dbx-app/mcp-server"
}
var stderr bytes.Buffer
cmd.Stderr = &stderr
if d.dataDir != "" {
if err := os.MkdirAll(d.dataDir, 0o755); err != nil {
return nil, fmt.Errorf("create dbx data dir: %w", err)
}
cmd.Env = append(os.Environ(), "DBX_DATA_DIR="+d.dataDir)
}
client := mcp.NewClient(&mcp.Implementation{Name: "primora", Version: "0.2.0"}, nil)
sess, err := client.Connect(ctx, &mcp.CommandTransport{Command: cmd}, nil)
if err != nil {
d.lastErr = fmt.Sprintf("dbx spawn failed: %v %s", err, stderr.String())
return nil, fmt.Errorf("%s", d.lastErr)
}
d.session = sess
d.cmd = cmd
d.registered = map[string]bool{}
d.lastErr = ""
return sess, nil
}
func (d *Client) reset() {
d.mu.Lock()
defer d.mu.Unlock()
if d.session != nil {
d.session.Close()
d.session = nil
}
if d.cmd != nil && d.cmd.Process != nil {
d.cmd.Process.Kill()
}
d.cmd = nil
d.registered = map[string]bool{}
}
// call invokes a DBX tool, transparently respawning the subprocess once on failure.
func (d *Client) call(ctx context.Context, tool string, args map[string]any) (string, error) {
sess, err := d.connect(ctx)
if err != nil {
return "", err
}
res, err := sess.CallTool(ctx, &mcp.CallToolParams{Name: tool, Arguments: args})
if err != nil {
d.reset()
sess, err2 := d.connect(ctx)
if err2 != nil {
return "", err
}
res, err = sess.CallTool(ctx, &mcp.CallToolParams{Name: tool, Arguments: args})
if err != nil {
return "", err
}
}
if res.IsError {
var msg string
for _, c := range res.Content {
if tc, ok := c.(*mcp.TextContent); ok {
msg += tc.Text
}
}
if msg == "" {
msg = "dbx tool error"
}
return "", fmt.Errorf("%s", msg)
}
var texts []string
for _, c := range res.Content {
if tc, ok := c.(*mcp.TextContent); ok {
texts = append(texts, tc.Text)
}
}
return strings.Join(texts, "\n"), nil
}
// Call runs a DBX tool with a bounded timeout and returns its text output.
func (d *Client) Call(ctx context.Context, tool string, args map[string]any) (string, error) {
ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
defer cancel()
return d.call(ctx, tool, args)
}
// EnsureConnection registers the connection in DBX storage when missing.
// Registration is cached per subprocess session and cleared on respawn.
func (d *Client) EnsureConnection(ctx context.Context, name string, args map[string]any) error {
d.mu.Lock()
if d.registered[name] {
d.mu.Unlock()
return nil
}
d.mu.Unlock()
_, err := d.Call(ctx, "dbx_add_connection", args)
if err == nil {
d.mu.Lock()
d.registered[name] = true
d.mu.Unlock()
return nil
}
// The add may have failed because the name is already in DBX storage —
// verify with a list before surfacing the error.
list, lerr := d.Call(ctx, "dbx_list_connections", nil)
if lerr == nil {
for _, row := range ParseConnectionRows(list) {
if row.Name == name {
d.mu.Lock()
d.registered[name] = true
d.mu.Unlock()
return nil
}
}
}
return err
}
// RemoveConnection drops the connection from DBX storage, best-effort.
func (d *Client) RemoveConnection(ctx context.Context, name string) {
_, _ = d.Call(ctx, "dbx_remove_connection", map[string]any{"connection_name": name})
d.mu.Lock()
delete(d.registered, name)
d.mu.Unlock()
}
// Status reports whether the DBX subprocess can be spawned.
func (d *Client) Status(ctx context.Context) map[string]any {
ctx, cancel := context.WithTimeout(ctx, 45*time.Second)
defer cancel()
if _, err := d.connect(ctx); err != nil {
return map[string]any{
"available": false,
"reason": err.Error(),
"install": "npm install -g @dbx-app/mcp-server (https://github.com/t8y2/dbx)",
}
}
d.mu.Lock()
cmd := d.cmdLine
d.mu.Unlock()
return map[string]any{"available": true, "command": cmd}
}
// Close terminates the subprocess.
func (d *Client) Close() { d.reset() }
// --- response normalization: DBX tools return markdown tables / bullet lists ---
func splitMDRow(line string) []string {
line = strings.TrimSpace(line)
line = strings.TrimPrefix(line, "|")
line = strings.TrimSuffix(line, "|")
parts := strings.Split(line, "|")
for i := range parts {
parts[i] = strings.TrimSpace(parts[i])
}
return parts
}
func isMDSeparator(line string) bool {
line = strings.TrimSpace(line)
if !strings.HasPrefix(line, "|") {
return false
}
for _, r := range line {
if r != '|' && r != '-' && r != ':' && r != ' ' {
return false
}
}
return true
}
type MDTable struct {
Columns []string `json:"columns"`
Rows [][]string `json:"rows"`
Note string `json:"note,omitempty"`
}
func ParseMDTable(text string) (*MDTable, string) {
var cols []string
var rows [][]string
var rest []string
for _, line := range strings.Split(text, "\n") {
trim := strings.TrimSpace(line)
switch {
case strings.HasPrefix(trim, "|") && len(cols) == 0:
cols = splitMDRow(trim)
case isMDSeparator(trim):
continue
case strings.HasPrefix(trim, "|"):
rows = append(rows, splitMDRow(trim))
default:
if trim != "" {
rest = append(rest, trim)
}
}
}
if len(cols) == 0 {
return nil, text
}
return &MDTable{Columns: cols, Rows: rows, Note: strings.Join(rest, "\n")}, text
}
type DBObject struct {
Name string `json:"name"`
Kind string `json:"kind"`
}
// ParseBullets handles "- name (KIND)" lists (dbx_list_tables, dbx_list_databases).
func ParseBullets(text string) []DBObject {
var out []DBObject
for _, line := range strings.Split(text, "\n") {
trim := strings.TrimSpace(line)
if !strings.HasPrefix(trim, "- ") {
continue
}
item := strings.TrimPrefix(trim, "- ")
kind := ""
if i := strings.LastIndex(item, " ("); i > 0 && strings.HasSuffix(item, ")") {
kind = item[i+2 : len(item)-1]
item = item[:i]
}
out = append(out, DBObject{Name: item, Kind: kind})
}
return out
}
type ConnectionRow struct {
ID string `json:"id"`
Name string `json:"name"`
Type string `json:"type"`
Host string `json:"host"`
Port string `json:"port"`
Database string `json:"database"`
Group string `json:"group"`
}
// ParseConnectionRows reads the `dbx_list_connections` markdown table.
func ParseConnectionRows(text string) []ConnectionRow {
t, _ := ParseMDTable(text)
if t == nil {
return nil
}
idx := map[string]int{}
for i, c := range t.Columns {
idx[strings.ToLower(c)] = i
}
get := func(row []string, col string) string {
if i, ok := idx[col]; ok && i < len(row) {
return row[i]
}
return ""
}
out := []ConnectionRow{}
for _, r := range t.Rows {
out = append(out, ConnectionRow{
ID: get(r, "id"), Name: get(r, "name"), Type: get(r, "type"),
Host: get(r, "host"), Port: get(r, "port"), Database: get(r, "database"),
Group: get(r, "group path"),
})
}
return out
}
+234
View File
@@ -0,0 +1,234 @@
package handlers
import (
"errors"
"net/http"
"strings"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/tdvorak/primora/apps/backend/internal/middleware"
"github.com/tdvorak/primora/apps/backend/internal/models"
apperrors "github.com/tdvorak/primora/apps/backend/internal/response"
"github.com/tdvorak/primora/apps/backend/internal/services"
)
func (h *HTTPHandler) dbxStatus(c *gin.Context) {
actor, ok := middleware.RequireActor(c)
if !ok {
return
}
result, err := h.Platform.DBXStatus(c.Request.Context(), actor)
if err != nil {
h.handleError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
func (h *HTTPHandler) listDBConnections(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
result, err := h.Platform.ListDBConnections(c.Request.Context(), actor, projectID)
if err != nil {
h.handleError(c, err)
return
}
c.JSON(http.StatusOK, gin.H{"items": result})
}
func (h *HTTPHandler) createDBConnection(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
var body services.CreateDBConnectionInput
if !h.bindAndValidate(c, &body) {
return
}
result, err := h.Platform.CreateDBConnection(c.Request.Context(), actor, projectID, body, middleware.RequestIDFromContext(c))
if err != nil {
if strings.Contains(err.Error(), "already exists") {
apperrors.Abort(c, http.StatusConflict, "name_conflict", err.Error())
return
}
h.handleError(c, err)
return
}
c.JSON(http.StatusCreated, result)
}
func (h *HTTPHandler) deleteDBConnection(c *gin.Context) {
actor, projectID, connectionID, ok := h.actorProjectConn(c)
if !ok {
return
}
if err := h.Platform.DeleteDBConnection(c.Request.Context(), actor, projectID, connectionID, middleware.RequestIDFromContext(c)); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
apperrors.Abort(c, http.StatusNotFound, "not_found", "connection not found")
return
}
if strings.Contains(err.Error(), "platform-managed") {
apperrors.Abort(c, http.StatusConflict, "managed_connection", err.Error())
return
}
h.handleError(c, err)
return
}
c.Status(http.StatusNoContent)
}
func (h *HTTPHandler) testDBConnection(c *gin.Context) {
actor, projectID, connectionID, ok := h.actorProjectConn(c)
if !ok {
return
}
result, err := h.Platform.TestDBConnection(c.Request.Context(), actor, projectID, connectionID)
if err != nil {
h.dbxError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
func (h *HTTPHandler) dbxDatabases(c *gin.Context) {
actor, projectID, connectionID, ok := h.actorProjectConn(c)
if !ok {
return
}
result, err := h.Platform.ListDBXDatabases(c.Request.Context(), actor, projectID, connectionID)
if err != nil {
h.dbxError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
func (h *HTTPHandler) dbxTables(c *gin.Context) {
actor, projectID, connectionID, ok := h.actorProjectConn(c)
if !ok {
return
}
result, err := h.Platform.ListDBXTables(c.Request.Context(), actor, projectID, connectionID,
strings.TrimSpace(c.Query("database")), strings.TrimSpace(c.Query("schema")))
if err != nil {
h.dbxError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
func (h *HTTPHandler) dbxDescribe(c *gin.Context) {
actor, projectID, connectionID, ok := h.actorProjectConn(c)
if !ok {
return
}
result, err := h.Platform.DescribeDBXTable(c.Request.Context(), actor, projectID, connectionID,
strings.TrimSpace(c.Query("table")), strings.TrimSpace(c.Query("database")), strings.TrimSpace(c.Query("schema")))
if err != nil {
h.dbxError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
func (h *HTTPHandler) dbxSchema(c *gin.Context) {
actor, projectID, connectionID, ok := h.actorProjectConn(c)
if !ok {
return
}
result, err := h.Platform.DBXSchemaContext(c.Request.Context(), actor, projectID, connectionID,
strings.TrimSpace(c.Query("database")), strings.TrimSpace(c.Query("schema")))
if err != nil {
h.dbxError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
func (h *HTTPHandler) dbxQuery(c *gin.Context) {
actor, projectID, connectionID, ok := h.actorProjectConn(c)
if !ok {
return
}
var body struct {
SQL string `json:"sql" validate:"required"`
Database string `json:"database"`
}
if !h.bindAndValidate(c, &body) {
return
}
result, err := h.Platform.ExecuteDBXQuery(c.Request.Context(), actor, projectID, connectionID, body.SQL, strings.TrimSpace(body.Database))
if err != nil {
h.dbxError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
func (h *HTTPHandler) dbxRedis(c *gin.Context) {
actor, projectID, connectionID, ok := h.actorProjectConn(c)
if !ok {
return
}
var body struct {
Command string `json:"command" validate:"required"`
DB *int `json:"db"`
}
if !h.bindAndValidate(c, &body) {
return
}
result, err := h.Platform.ExecuteDBXRedis(c.Request.Context(), actor, projectID, connectionID, body.Command, body.DB)
if err != nil {
h.dbxError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
// actorAndProject resolves the actor and :projectID param.
func (h *HTTPHandler) actorAndProject(c *gin.Context) (*models.Actor, uuid.UUID, bool) {
actor, ok := middleware.RequireActor(c)
if !ok {
return nil, uuid.Nil, false
}
projectID, ok := parseUUIDParam(c, "projectID")
if !ok {
return nil, uuid.Nil, false
}
return actor, projectID, true
}
// actorProjectConn resolves the actor, :projectID and :connectionID params.
func (h *HTTPHandler) actorProjectConn(c *gin.Context) (*models.Actor, uuid.UUID, uuid.UUID, bool) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return nil, uuid.Nil, uuid.Nil, false
}
connectionID, ok := parseUUIDParam(c, "connectionID")
if !ok {
return nil, uuid.Nil, uuid.Nil, false
}
return actor, projectID, connectionID, true
}
// dbxError maps DBX subprocess/tool failures to 502, not-found rows to 404.
func (h *HTTPHandler) dbxError(c *gin.Context, err error) {
if errors.Is(err, pgx.ErrNoRows) {
apperrors.Abort(c, http.StatusNotFound, "not_found", "connection not found")
return
}
if strings.Contains(err.Error(), "access denied") || strings.Contains(err.Error(), "insufficient") || strings.Contains(err.Error(), "authentication required") {
h.handleError(c, err)
return
}
if strings.Contains(err.Error(), "table required") {
apperrors.Abort(c, http.StatusBadRequest, "bad_request", err.Error())
return
}
apperrors.Abort(c, http.StatusBadGateway, "dbx_error", err.Error())
}
+11
View File
@@ -70,6 +70,17 @@ func (h *HTTPHandler) Register(router *gin.Engine) {
api.GET("/collections/:collectionID/documents/:documentID", h.getDocument)
api.PATCH("/collections/:collectionID/documents/:documentID", h.updateDocument)
api.DELETE("/collections/:collectionID/documents/:documentID", h.deleteDocument)
api.GET("/db/status", h.dbxStatus)
api.GET("/projects/:projectID/db-connections", h.listDBConnections)
api.POST("/projects/:projectID/db-connections", h.createDBConnection)
api.DELETE("/projects/:projectID/db-connections/:connectionID", h.deleteDBConnection)
api.POST("/projects/:projectID/db-connections/:connectionID/test", h.testDBConnection)
api.GET("/projects/:projectID/db-connections/:connectionID/databases", h.dbxDatabases)
api.GET("/projects/:projectID/db-connections/:connectionID/tables", h.dbxTables)
api.GET("/projects/:projectID/db-connections/:connectionID/describe", h.dbxDescribe)
api.GET("/projects/:projectID/db-connections/:connectionID/schema", h.dbxSchema)
api.POST("/projects/:projectID/db-connections/:connectionID/query", h.dbxQuery)
api.POST("/projects/:projectID/db-connections/:connectionID/redis", h.dbxRedis)
api.GET("/buckets/:bucketID/objects", h.listObjects)
api.POST("/buckets/:bucketID/objects", h.uploadObject)
api.POST("/buckets/:bucketID/object-copies", h.copyObject)
+414
View File
@@ -0,0 +1,414 @@
package services
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
db "github.com/tdvorak/primora/apps/backend/internal/database/db"
"github.com/tdvorak/primora/apps/backend/internal/dbx"
"github.com/tdvorak/primora/apps/backend/internal/models"
)
// DBConnectionConfig holds the server-side credential payload for a saved
// connection. It is stored in core.db_connections.config and never serialized
// to the client — see DBConnectionSummary.
type DBConnectionConfig struct {
Host string `json:"host"`
Port *int `json:"port,omitempty"`
Database string `json:"database,omitempty"`
Username string `json:"username,omitempty"`
Password string `json:"password,omitempty"`
SSL *bool `json:"ssl,omitempty"`
DriverProfile string `json:"driver_profile,omitempty"`
}
// ManagedDBSeed describes a first-party connection auto-registered on every
// project (the platform's own Postgres and Dragonfly).
type ManagedDBSeed struct {
Name string
DBType string
Config DBConnectionConfig
}
type CreateDBConnectionInput struct {
Name string `json:"name" validate:"required,min=1"`
DBType string `json:"db_type" validate:"required,oneof=postgres mysql sqlite redis mongodb sqlserver duckdb clickhouse cassandra"`
Host string `json:"host" validate:"required"`
Port *int `json:"port" validate:"omitempty,min=1,max=65535"`
Database string `json:"database"`
Username string `json:"username"`
Password string `json:"password"`
SSL *bool `json:"ssl"`
DriverProfile string `json:"driver_profile"`
}
// DBConnectionSummary is the client-facing shape — config minus secrets.
type DBConnectionSummary struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
DBType string `json:"db_type"`
Host string `json:"host"`
Port *int `json:"port"`
Database string `json:"database"`
Username string `json:"username"`
SSL *bool `json:"ssl"`
IsManaged bool `json:"is_managed"`
HasPassword bool `json:"has_password"`
}
func parseDBConnectionConfig(raw []byte) DBConnectionConfig {
var cfg DBConnectionConfig
_ = json.Unmarshal(raw, &cfg)
return cfg
}
func toDBConnectionSummary(row db.CoreDbConnection) DBConnectionSummary {
cfg := parseDBConnectionConfig(row.Config)
return DBConnectionSummary{
ID: row.ID,
ProjectID: row.ProjectID,
Name: row.Name,
DBType: row.DbType,
Host: cfg.Host,
Port: cfg.Port,
Database: cfg.Database,
Username: cfg.Username,
SSL: cfg.SSL,
IsManaged: row.IsManaged,
HasPassword: cfg.Password != "",
}
}
// dbxConnectionName derives the DBX-internal name. The project prefix keeps
// same-named connections in different projects from clobbering each other in
// DBX's shared storage.
func dbxConnectionName(projectID uuid.UUID, name string) string {
short := strings.ReplaceAll(projectID.String(), "-", "")[:8]
return "prj_" + short + "_" + name
}
func (s *PlatformService) dbxArgsFor(row db.CoreDbConnection) map[string]any {
cfg := parseDBConnectionConfig(row.Config)
args := map[string]any{
"name": dbxConnectionName(row.ProjectID, row.Name),
"db_type": row.DbType,
"host": cfg.Host,
}
if cfg.Port != nil {
args["port"] = *cfg.Port
}
if cfg.Database != "" {
args["database"] = cfg.Database
}
if cfg.Username != "" {
args["username"] = cfg.Username
}
if cfg.Password != "" {
args["password"] = cfg.Password
}
if cfg.SSL != nil {
args["ssl"] = *cfg.SSL
}
if cfg.DriverProfile != "" {
args["driver_profile"] = cfg.DriverProfile
}
return args
}
// seedManagedConnections registers the platform's own Postgres and Dragonfly
// on the project. Runs on list so existing projects pick them up too.
func (s *PlatformService) seedManagedConnections(ctx context.Context, projectID uuid.UUID) {
for _, seed := range s.managedDBSeeds {
_, err := s.repo.Queries().GetDBConnectionByName(ctx, db.GetDBConnectionByNameParams{
ProjectID: projectID,
Name: seed.Name,
})
if err == nil {
continue
}
if !errors.Is(err, pgx.ErrNoRows) {
continue
}
cfgBytes, _ := json.Marshal(seed.Config)
_, _ = s.repo.Queries().CreateDBConnection(ctx, db.CreateDBConnectionParams{
ProjectID: projectID,
Name: seed.Name,
DbType: seed.DBType,
Config: cfgBytes,
IsManaged: true,
})
}
}
func (s *PlatformService) DBXStatus(ctx context.Context, actor *models.Actor) (map[string]any, error) {
if actor == nil {
return nil, errors.New("authentication required")
}
if s.dbx == nil {
return map[string]any{"available": false, "reason": "dbx not configured"}, nil
}
return s.dbx.Status(ctx), nil
}
func (s *PlatformService) ListDBConnections(ctx context.Context, actor *models.Actor, projectID uuid.UUID) ([]DBConnectionSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
s.seedManagedConnections(ctx, projectID)
rows, err := s.repo.Queries().ListDBConnections(ctx, projectID)
if err != nil {
return nil, err
}
out := make([]DBConnectionSummary, 0, len(rows))
for _, row := range rows {
out = append(out, toDBConnectionSummary(row))
}
return out, nil
}
func (s *PlatformService) CreateDBConnection(ctx context.Context, actor *models.Actor, projectID uuid.UUID, input CreateDBConnectionInput, requestID string) (DBConnectionSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return DBConnectionSummary{}, err
}
name := normalizeSlug(input.Name)
if name == "" {
return DBConnectionSummary{}, errors.New("name must contain letters or digits")
}
cfg := DBConnectionConfig{
Host: strings.TrimSpace(input.Host),
Port: input.Port,
Database: strings.TrimSpace(input.Database),
Username: strings.TrimSpace(input.Username),
Password: input.Password,
SSL: input.SSL,
DriverProfile: strings.TrimSpace(input.DriverProfile),
}
cfgBytes, _ := json.Marshal(cfg)
var createdBy pgtype.UUID
if actor.UserID != nil {
createdBy = pgtype.UUID{Bytes: *actor.UserID, Valid: true}
}
row, err := s.repo.Queries().CreateDBConnection(ctx, db.CreateDBConnectionParams{
ProjectID: projectID,
Name: name,
DbType: input.DBType,
Config: cfgBytes,
CreatedByUserID: createdBy,
})
if err != nil {
if strings.Contains(err.Error(), "db_connections_project_id_name_key") {
return DBConnectionSummary{}, fmt.Errorf("a connection named %s already exists", name)
}
return DBConnectionSummary{}, err
}
project, err := s.repo.Queries().GetProjectByID(ctx, projectID)
if err == nil {
_, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "db_connection.created", "db_connection", row.ID.String(), map[string]any{
"name": row.Name, "db_type": row.DbType,
}))
}
return toDBConnectionSummary(row), nil
}
func (s *PlatformService) DeleteDBConnection(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID, requestID string) error {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return err
}
row, err := s.repo.Queries().GetDBConnectionByID(ctx, connectionID)
if err != nil {
return err
}
if row.ProjectID != projectID {
return fmt.Errorf("connection access denied")
}
if row.IsManaged {
return fmt.Errorf("platform-managed connections cannot be removed")
}
if _, err := s.repo.Queries().DeleteDBConnection(ctx, db.DeleteDBConnectionParams{
ID: connectionID,
ProjectID: projectID,
}); err != nil {
return err
}
if s.dbx != nil {
s.dbx.RemoveConnection(ctx, dbxConnectionName(projectID, row.Name))
}
project, _ := s.repo.Queries().GetProjectByID(ctx, projectID)
_, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "db_connection.deleted", "db_connection", connectionID.String(), map[string]any{
"name": row.Name,
}))
return nil
}
// connForTool resolves the connection row, checks access, and registers it
// with the DBX subprocess. Write tools require admin/developer.
func (s *PlatformService) connForTool(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID, write bool) (db.CoreDbConnection, error) {
roles := []string{"admin", "developer", "viewer"}
if write {
roles = []string{"admin", "developer"}
}
if err := s.requireProjectRole(ctx, actor, projectID, roles...); err != nil {
return db.CoreDbConnection{}, err
}
row, err := s.repo.Queries().GetDBConnectionByID(ctx, connectionID)
if err != nil {
return db.CoreDbConnection{}, err
}
if row.ProjectID != projectID {
return db.CoreDbConnection{}, fmt.Errorf("connection access denied")
}
if s.dbx == nil {
return db.CoreDbConnection{}, errors.New("dbx not configured")
}
if err := s.dbx.EnsureConnection(ctx, dbxConnectionName(projectID, row.Name), s.dbxArgsFor(row)); err != nil {
return db.CoreDbConnection{}, err
}
return row, nil
}
func (s *PlatformService) TestDBConnection(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID) (map[string]any, error) {
row, err := s.connForTool(ctx, actor, projectID, connectionID, true)
if err != nil {
return nil, err
}
name := dbxConnectionName(projectID, row.Name)
tool, args := "dbx_list_databases", map[string]any{"connection_name": name}
if row.DbType == "redis" {
tool, args = "dbx_execute_redis_command", map[string]any{"connection_name": name, "command": "PING"}
}
if _, err := s.dbx.Call(ctx, tool, args); err != nil {
return map[string]any{"ok": false, "error": err.Error()}, nil
}
return map[string]any{"ok": true}, nil
}
func (s *PlatformService) ListDBXDatabases(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID) (map[string]any, error) {
row, err := s.connForTool(ctx, actor, projectID, connectionID, false)
if err != nil {
return nil, err
}
text, err := s.dbx.Call(ctx, "dbx_list_databases", map[string]any{"connection_name": dbxConnectionName(projectID, row.Name)})
if err != nil {
return nil, err
}
if objs := dbx.ParseBullets(text); objs != nil {
names := make([]string, len(objs))
for i, o := range objs {
names[i] = o.Name
}
return map[string]any{"databases": names}, nil
}
return map[string]any{"databases": []string{}, "raw": text}, nil
}
func (s *PlatformService) ListDBXTables(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID, database, schema string) (map[string]any, error) {
row, err := s.connForTool(ctx, actor, projectID, connectionID, false)
if err != nil {
return nil, err
}
args := map[string]any{"connection_name": dbxConnectionName(projectID, row.Name)}
if database != "" {
args["database"] = database
}
if schema != "" {
args["schema"] = schema
}
text, err := s.dbx.Call(ctx, "dbx_list_tables", args)
if err != nil {
return nil, err
}
if objs := dbx.ParseBullets(text); objs != nil {
return map[string]any{"tables": objs}, nil
}
if t, _ := dbx.ParseMDTable(text); t != nil {
return map[string]any{"table": t}, nil
}
return map[string]any{"tables": []dbx.DBObject{}, "raw": text}, nil
}
func (s *PlatformService) DescribeDBXTable(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID, table, database, schema string) (map[string]any, error) {
if table == "" {
return nil, errors.New("table required")
}
row, err := s.connForTool(ctx, actor, projectID, connectionID, false)
if err != nil {
return nil, err
}
args := map[string]any{"connection_name": dbxConnectionName(projectID, row.Name), "table": table}
if database != "" {
args["database"] = database
}
if schema != "" {
args["schema"] = schema
}
text, err := s.dbx.Call(ctx, "dbx_describe_table", args)
if err != nil {
return nil, err
}
if t, _ := dbx.ParseMDTable(text); t != nil {
return map[string]any{"table": t}, nil
}
return map[string]any{"raw": text}, nil
}
func (s *PlatformService) DBXSchemaContext(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID, database, schema string) (map[string]any, error) {
row, err := s.connForTool(ctx, actor, projectID, connectionID, false)
if err != nil {
return nil, err
}
args := map[string]any{"connection_name": dbxConnectionName(projectID, row.Name)}
if database != "" {
args["database"] = database
}
if schema != "" {
args["schema"] = schema
}
text, err := s.dbx.Call(ctx, "dbx_get_schema_context", args)
if err != nil {
return nil, err
}
return map[string]any{"context": text}, nil
}
func (s *PlatformService) ExecuteDBXQuery(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID, sql, database string) (map[string]any, error) {
row, err := s.connForTool(ctx, actor, projectID, connectionID, true)
if err != nil {
return nil, err
}
args := map[string]any{"connection_name": dbxConnectionName(projectID, row.Name), "sql": sql}
if database != "" {
args["database"] = database
}
text, err := s.dbx.Call(ctx, "dbx_execute_query", args)
if err != nil {
return nil, err
}
if t, _ := dbx.ParseMDTable(text); t != nil {
return map[string]any{"table": t}, nil
}
return map[string]any{"raw": text}, nil
}
func (s *PlatformService) ExecuteDBXRedis(ctx context.Context, actor *models.Actor, projectID, connectionID uuid.UUID, command string, dbIndex *int) (map[string]any, error) {
row, err := s.connForTool(ctx, actor, projectID, connectionID, true)
if err != nil {
return nil, err
}
args := map[string]any{"connection_name": dbxConnectionName(projectID, row.Name), "command": command}
if dbIndex != nil {
args["db"] = *dbIndex
}
text, err := s.dbx.Call(ctx, "dbx_execute_redis_command", args)
if err != nil {
return nil, err
}
return map[string]any{"output": text}, nil
}
@@ -19,16 +19,19 @@ import (
"github.com/jackc/pgx/v5/pgtype"
db "github.com/tdvorak/primora/apps/backend/internal/database/db"
"github.com/tdvorak/primora/apps/backend/internal/dbx"
"github.com/tdvorak/primora/apps/backend/internal/models"
"github.com/tdvorak/primora/apps/backend/internal/repositories"
"github.com/tdvorak/primora/apps/backend/internal/storage"
)
type PlatformService struct {
repo *repositories.CoreRepository
store *storage.LocalStore
mailer *Mailer
publicURL string
repo *repositories.CoreRepository
store *storage.LocalStore
mailer *Mailer
publicURL string
dbx *dbx.Client
managedDBSeeds []ManagedDBSeed
}
type BootstrapInput struct {
@@ -206,8 +209,8 @@ type InvitationSummary struct {
Status string `json:"status"`
}
func NewPlatformService(repo *repositories.CoreRepository, store *storage.LocalStore, mailer *Mailer, publicURL string) *PlatformService {
return &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL}
func NewPlatformService(repo *repositories.CoreRepository, store *storage.LocalStore, mailer *Mailer, publicURL string, dbxClient *dbx.Client, managedDBSeeds []ManagedDBSeed) *PlatformService {
return &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, managedDBSeeds: managedDBSeeds}
}
func (s *PlatformService) Me(ctx context.Context, actor *models.Actor) (PlatformSummary, error) {