feat(control-center): P0 skeleton — standalone server, pgvector compose, goose schema, token auth

This commit is contained in:
weichen
2026-09-17 08:38:50 +00:00
parent cf805a73a7
commit 2be43efddd
8 changed files with 741 additions and 0 deletions
+149
View File
@@ -0,0 +1,149 @@
// Control Center: personal project control plane, independent of the product
// system. Subcommands: serve, migrate, register-agent.
package main
import (
"context"
"crypto/rand"
"encoding/hex"
"flag"
"fmt"
"log/slog"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"chat-agent-system/internal/controlcenter/config"
"chat-agent-system/internal/controlcenter/server"
"chat-agent-system/internal/controlcenter/store"
)
func main() {
if len(os.Args) < 2 {
usage()
os.Exit(2)
}
var err error
switch os.Args[1] {
case "serve":
err = cmdServe(os.Args[2:])
case "migrate":
err = cmdMigrate(os.Args[2:])
case "register-agent":
err = cmdRegisterAgent(os.Args[2:])
default:
usage()
os.Exit(2)
}
if err != nil {
slog.Error("control-center", "err", err)
os.Exit(1)
}
}
func usage() {
fmt.Fprintln(os.Stderr, `control-center — personal project control plane
usage:
control-center serve run HTTP server (CC_HTTP_ADDR, default 127.0.0.1:8790)
control-center migrate apply goose migrations only
control-center register-agent -name X [-kind K] [-token T] create/rotate an agent token`)
}
func cmdServe(args []string) error {
fs := flag.NewFlagSet("serve", flag.ContinueOnError)
if err := fs.Parse(args); err != nil {
return err
}
cfg, err := config.Load()
if err != nil {
return err
}
ctx := context.Background()
if err := store.Migrate(ctx, cfg.DBDSN, cfg.MigrationsDir); err != nil {
return err
}
st, err := store.Connect(ctx, cfg.DBDSN)
if err != nil {
return err
}
defer st.Close()
if cfg.ConsoleToken == "" {
cfg.ConsoleToken = randomToken()
slog.Warn("CC_CONSOLE_TOKEN not set; generated ephemeral console token")
}
srv := server.New(cfg, st)
httpSrv := &http.Server{Addr: cfg.HTTPAddr, Handler: srv.Handler(), ReadHeaderTimeout: 10 * time.Second}
go func() {
sig := make(chan os.Signal, 1)
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
<-sig
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = httpSrv.Shutdown(shutdownCtx)
}()
slog.Info("control-center listening", "addr", cfg.HTTPAddr)
fmt.Printf("console token: %s\n", cfg.ConsoleToken)
if err := httpSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
return err
}
return nil
}
func cmdMigrate(args []string) error {
fs := flag.NewFlagSet("migrate", flag.ContinueOnError)
if err := fs.Parse(args); err != nil {
return err
}
cfg, err := config.Load()
if err != nil {
return err
}
return store.Migrate(context.Background(), cfg.DBDSN, cfg.MigrationsDir)
}
func cmdRegisterAgent(args []string) error {
fs := flag.NewFlagSet("register-agent", flag.ContinueOnError)
name := fs.String("name", "", "agent name (required)")
kind := fs.String("kind", "generic", "agent kind, e.g. cursor / kimi / pi")
token := fs.String("token", "", "plaintext token; generated when empty")
if err := fs.Parse(args); err != nil {
return err
}
if *name == "" {
return fmt.Errorf("-name is required")
}
cfg, err := config.Load()
if err != nil {
return err
}
ctx := context.Background()
if err := store.Migrate(ctx, cfg.DBDSN, cfg.MigrationsDir); err != nil {
return err
}
st, err := store.Connect(ctx, cfg.DBDSN)
if err != nil {
return err
}
defer st.Close()
if *token == "" {
*token = randomToken()
}
if err := server.RegisterAgent(ctx, st, *name, *kind, *token); err != nil {
return err
}
fmt.Printf("agent %q registered. token (shown once): %s\n", *name, *token)
return nil
}
func randomToken() string {
b := make([]byte, 24)
if _, err := rand.Read(b); err != nil {
panic(err)
}
return hex.EncodeToString(b)
}
+30
View File
@@ -0,0 +1,30 @@
name: control-center
services:
db:
image: pgvector/pgvector:pg16
container_name: control-center-db
restart: unless-stopped
environment:
POSTGRES_USER: ${CC_DB_USER:-cc}
POSTGRES_PASSWORD: ${CC_DB_PASSWORD:-cc}
POSTGRES_DB: ${CC_DB_NAME:-control_center}
ports:
- "${CC_DB_PORT:-5440}:5432"
volumes:
- cc-db-data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER:-cc} -d $${POSTGRES_DB:-control_center}"]
interval: 5s
timeout: 3s
retries: 12
volumes:
cc-db-data:
# Docker's predefined pool is exhausted on this host; pin an explicit subnet.
networks:
default:
ipam:
config:
- subnet: 172.32.0.0/24
+202
View File
@@ -0,0 +1,202 @@
-- +goose Up
-- Control Center core schema: plans, tasks, 8-stage pipeline, agent collaboration,
-- document/artifact vector store, audit findings, snapshots.
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE cc_plans (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
slug text NOT NULL UNIQUE,
title text NOT NULL,
horizon text NOT NULL DEFAULT 'short' CHECK (horizon IN ('short','long','backlog')),
status text NOT NULL DEFAULT 'planned' CHECK (status IN ('planned','in_progress','done','cancelled')),
summary_md text NOT NULL DEFAULT '',
doc_path text NOT NULL DEFAULT '',
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE cc_agents (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
name text NOT NULL UNIQUE,
kind text NOT NULL DEFAULT 'generic',
token_hash text NOT NULL UNIQUE,
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE cc_tasks (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
slug text NOT NULL UNIQUE,
plan_id uuid REFERENCES cc_plans(id) ON DELETE SET NULL,
title text NOT NULL,
kind text NOT NULL DEFAULT 'task' CHECK (kind IN ('task','proposal')),
status text NOT NULL DEFAULT 'planned'
CHECK (status IN ('planned','in_progress','blocked_pending_decision','done','cancelled')),
stage text NOT NULL DEFAULT 'proposal'
CHECK (stage IN ('proposal','analysis','design','review','implementation','testing','evaluation','confirmation')),
stage_state text NOT NULL DEFAULT 'idle'
CHECK (stage_state IN ('idle','assigned','in_progress','awaiting_review','changes_requested','approved')),
testing_required boolean NOT NULL DEFAULT true,
priority int NOT NULL DEFAULT 0,
owner text NOT NULL DEFAULT '',
agent_kind text NOT NULL DEFAULT '',
mutex text NOT NULL DEFAULT '',
branch text NOT NULL DEFAULT '',
worktree text NOT NULL DEFAULT '',
instruction_md text NOT NULL DEFAULT '',
doc_path text NOT NULL DEFAULT '',
assigned_agent text NOT NULL DEFAULT '',
lease_expires_at timestamptz,
heartbeat_at timestamptz,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX cc_tasks_status_idx ON cc_tasks (status);
CREATE INDEX cc_tasks_stage_idx ON cc_tasks (stage, stage_state);
CREATE INDEX cc_tasks_plan_idx ON cc_tasks (plan_id);
CREATE TABLE cc_task_events (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
task_id uuid NOT NULL REFERENCES cc_tasks(id) ON DELETE CASCADE,
agent text NOT NULL DEFAULT '',
phase text NOT NULL CHECK (phase IN ('start','plan','progress','blocked','report','done','advance-request','decision','system')),
message text NOT NULL DEFAULT '',
payload jsonb NOT NULL DEFAULT '{}'::jsonb,
ts timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX cc_task_events_task_idx ON cc_task_events (task_id, id);
CREATE TABLE cc_task_reports (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
task_id uuid NOT NULL REFERENCES cc_tasks(id) ON DELETE CASCADE,
agent text NOT NULL DEFAULT '',
kind text NOT NULL DEFAULT 'result' CHECK (kind IN ('result','test-report','handoff')),
content_md text NOT NULL,
links jsonb NOT NULL DEFAULT '[]'::jsonb,
ts timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX cc_task_reports_task_idx ON cc_task_reports (task_id, id);
CREATE TABLE cc_stage_artifacts (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
task_id uuid NOT NULL REFERENCES cc_tasks(id) ON DELETE CASCADE,
stage text NOT NULL,
kind text NOT NULL DEFAULT 'note',
title text NOT NULL DEFAULT '',
content_md text NOT NULL,
source_type text NOT NULL DEFAULT 'agent' CHECK (source_type IN ('agent','human','external')),
source_ref text NOT NULL DEFAULT '',
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX cc_stage_artifacts_task_idx ON cc_stage_artifacts (task_id, stage);
CREATE TABLE cc_stage_transitions (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
task_id uuid NOT NULL REFERENCES cc_tasks(id) ON DELETE CASCADE,
from_stage text NOT NULL,
to_stage text NOT NULL,
actor text NOT NULL,
approved_by text NOT NULL,
note text NOT NULL DEFAULT '',
ts timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX cc_stage_transitions_task_idx ON cc_stage_transitions (task_id, id);
CREATE TABLE cc_stage_settings (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
scope text NOT NULL DEFAULT 'global' CHECK (scope IN ('global','task')),
task_id uuid REFERENCES cc_tasks(id) ON DELETE CASCADE,
stage text NOT NULL,
model text NOT NULL DEFAULT '',
executor text NOT NULL DEFAULT 'agent' CHECK (executor IN ('agent','human','external')),
updated_at timestamptz NOT NULL DEFAULT now(),
UNIQUE (scope, task_id, stage)
);
CREATE TABLE cc_decisions (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
task_id uuid NOT NULL REFERENCES cc_tasks(id) ON DELETE CASCADE,
stage text NOT NULL,
question text NOT NULL,
context_md text NOT NULL DEFAULT '',
suggestions jsonb NOT NULL DEFAULT '[]'::jsonb,
status text NOT NULL DEFAULT 'open' CHECK (status IN ('open','resolved')),
resolution text NOT NULL DEFAULT '',
decided_by text NOT NULL DEFAULT '',
created_at timestamptz NOT NULL DEFAULT now(),
resolved_at timestamptz
);
CREATE INDEX cc_decisions_open_idx ON cc_decisions (status) WHERE status = 'open';
CREATE TABLE cc_documents (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
path text NOT NULL UNIQUE,
title text NOT NULL DEFAULT '',
taxonomy text NOT NULL DEFAULT '',
frontmatter jsonb NOT NULL DEFAULT '{}'::jsonb,
sha256 text NOT NULL,
synced_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE cc_doc_chunks (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
doc_id uuid REFERENCES cc_documents(id) ON DELETE CASCADE,
artifact_id uuid REFERENCES cc_stage_artifacts(id) ON DELETE CASCADE,
seq int NOT NULL,
heading text NOT NULL DEFAULT '',
content text NOT NULL,
chunk_hash text NOT NULL UNIQUE,
embedding vector(1024),
tsv tsvector GENERATED ALWAYS AS (to_tsvector('simple', content)) STORED,
created_at timestamptz NOT NULL DEFAULT now(),
CHECK (doc_id IS NOT NULL OR artifact_id IS NOT NULL)
);
CREATE INDEX cc_doc_chunks_embedding_hnsw ON cc_doc_chunks USING hnsw (embedding vector_cosine_ops);
CREATE INDEX cc_doc_chunks_tsv_idx ON cc_doc_chunks USING gin (tsv);
CREATE TABLE cc_audit_runs (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
kind text NOT NULL DEFAULT 'daily',
status text NOT NULL DEFAULT 'running' CHECK (status IN ('running','completed','failed')),
summary text NOT NULL DEFAULT '',
error text NOT NULL DEFAULT '',
started_at timestamptz NOT NULL DEFAULT now(),
finished_at timestamptz
);
CREATE TABLE cc_audit_findings (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
run_id bigint NOT NULL REFERENCES cc_audit_runs(id) ON DELETE CASCADE,
severity text NOT NULL DEFAULT 'medium' CHECK (severity IN ('low','medium','high','critical')),
kind text NOT NULL DEFAULT 'gap' CHECK (kind IN ('bug','gap','unfinished','doc-drift')),
title text NOT NULL,
detail_md text NOT NULL DEFAULT '',
refs jsonb NOT NULL DEFAULT '[]'::jsonb,
status text NOT NULL DEFAULT 'open' CHECK (status IN ('open','converted','dismissed')),
task_id uuid REFERENCES cc_tasks(id) ON DELETE SET NULL,
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX cc_audit_findings_status_idx ON cc_audit_findings (status);
CREATE TABLE cc_snapshots (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
revision text NOT NULL,
payload jsonb NOT NULL,
created_at timestamptz NOT NULL DEFAULT now()
);
-- +goose Down
DROP TABLE IF EXISTS cc_snapshots;
DROP TABLE IF EXISTS cc_audit_findings;
DROP TABLE IF EXISTS cc_audit_runs;
DROP TABLE IF EXISTS cc_doc_chunks;
DROP TABLE IF EXISTS cc_documents;
DROP TABLE IF EXISTS cc_decisions;
DROP TABLE IF EXISTS cc_stage_settings;
DROP TABLE IF EXISTS cc_stage_transitions;
DROP TABLE IF EXISTS cc_stage_artifacts;
DROP TABLE IF EXISTS cc_task_reports;
DROP TABLE IF EXISTS cc_task_events;
DROP TABLE IF EXISTS cc_tasks;
DROP TABLE IF EXISTS cc_agents;
DROP TABLE IF EXISTS cc_plans;
+117
View File
@@ -0,0 +1,117 @@
// Package config loads Control Center configuration: CC_* variables from the
// process environment, plus borrowed keys (embedding / LLM) from the system
// .env file when CC_* overrides are absent.
package config
import (
"bufio"
"fmt"
"os"
"path/filepath"
"strings"
)
type Config struct {
HTTPAddr string
DBDSN string
ConsoleToken string
RepoRoot string
MigrationsDir string
EmbeddingEndpoint string
EmbeddingModel string
EmbeddingAPIKey string
LLMEndpoint string
LLMModel string
LLMAPIKey string
CursorModel string
AuditEnabled bool
AuditDailyAt string // "HH:MM" local time
DocSyncSecond int
}
// Load reads CC_* from the process env and falls back to keys borrowed from
// the system .env file (CC_ENV_FILE, default <repoRoot>/.env).
func Load() (Config, error) {
repoRoot := getenv("CC_REPO_ROOT", ".")
abs, err := filepath.Abs(repoRoot)
if err != nil {
return Config{}, fmt.Errorf("resolve CC_REPO_ROOT: %w", err)
}
envFile := getenv("CC_ENV_FILE", filepath.Join(abs, ".env"))
borrowed := parseEnvFile(envFile) // missing file is fine; values stay empty
pick := func(ccKey, borrowedKey, def string) string {
if v := os.Getenv(ccKey); v != "" {
return v
}
if borrowedKey != "" {
if v := borrowed[borrowedKey]; v != "" {
return v
}
if v := os.Getenv(borrowedKey); v != "" {
return v
}
}
return def
}
cfg := Config{
HTTPAddr: getenv("CC_HTTP_ADDR", "127.0.0.1:8790"),
DBDSN: getenv("CC_DB_DSN", "postgres://cc:cc@127.0.0.1:5440/control_center?sslmode=disable"),
ConsoleToken: os.Getenv("CC_CONSOLE_TOKEN"),
RepoRoot: abs,
MigrationsDir: getenv("CC_MIGRATIONS_DIR", filepath.Join(abs, "deploy", "control-center", "goose")),
EmbeddingEndpoint: pick("CC_EMBEDDING_ENDPOINT_URL", "EMBEDDING_ENDPOINT_URL", ""),
EmbeddingModel: pick("CC_EMBEDDING_MODEL", "EMBEDDING_MODEL", "bge-m3"),
EmbeddingAPIKey: pick("CC_EMBEDDING_API_KEY", "EMBEDDING_API_KEY", ""),
LLMEndpoint: pick("CC_LLM_ENDPOINT_URL", "MINIMAX_ENDPOINT_URL", ""),
LLMModel: pick("CC_LLM_MODEL", "MINIMAX_MODEL_ID", ""),
LLMAPIKey: pick("CC_LLM_API_KEY", "MINIMAX_API_KEY", ""),
CursorModel: pick("CC_CURSOR_MODEL", "CURSOR_MODEL", ""),
AuditEnabled: getenv("CC_AUDIT_ENABLED", "true") == "true",
AuditDailyAt: getenv("CC_AUDIT_DAILY_AT", "03:30"),
DocSyncSecond: 60,
}
return cfg, nil
}
func getenv(key, def string) string {
if v := os.Getenv(key); v != "" {
return v
}
return def
}
// parseEnvFile reads KEY=VALUE lines; comments and blank lines are skipped.
// It never fails: an unreadable file yields an empty map.
func parseEnvFile(path string) map[string]string {
out := map[string]string{}
f, err := os.Open(path)
if err != nil {
return out
}
defer f.Close()
sc := bufio.NewScanner(f)
sc.Buffer(make([]byte, 64*1024), 1024*1024)
for sc.Scan() {
line := strings.TrimSpace(sc.Text())
if line == "" || strings.HasPrefix(line, "#") {
continue
}
k, v, ok := strings.Cut(line, "=")
if !ok {
continue
}
out[strings.TrimSpace(k)] = strings.Trim(strings.TrimSpace(v), `"'`)
}
return out
}
+34
View File
@@ -0,0 +1,34 @@
package server
import (
"context"
"chat-agent-system/internal/controlcenter/store"
)
// AgentAuthenticator resolves bearer tokens against cc_agents.token_hash,
// with a small in-memory cache populated lazily.
type AgentAuthenticator struct {
st *store.Store
}
// Authenticate returns the agent name when the token is valid.
func (a *AgentAuthenticator) Authenticate(ctx context.Context, token string) (string, bool) {
var name string
err := a.st.Pool.QueryRow(ctx,
`SELECT name FROM cc_agents WHERE token_hash = $1`, TokenHash(token)).Scan(&name)
if err != nil {
return "", false
}
return name, true
}
// RegisterAgent creates (or rotates) an agent token; returns the plaintext
// token exactly once.
func RegisterAgent(ctx context.Context, st *store.Store, name, kind, token string) error {
_, err := st.Pool.Exec(ctx,
`INSERT INTO cc_agents (name, kind, token_hash) VALUES ($1,$2,$3)
ON CONFLICT (name) DO UPDATE SET kind = EXCLUDED.kind, token_hash = EXCLUDED.token_hash`,
name, kind, TokenHash(token))
return err
}
+121
View File
@@ -0,0 +1,121 @@
// Package server exposes the Control Center HTTP API. Loopback-first, token
// authenticated: console token for the human UI, per-agent bearer tokens for
// /api/v1/agent/*.
package server
import (
"context"
"crypto/sha256"
"crypto/subtle"
"encoding/hex"
"encoding/json"
"log/slog"
"net/http"
"strings"
"chat-agent-system/internal/controlcenter/config"
"chat-agent-system/internal/controlcenter/store"
)
type ctxKey string
const ctxAgent ctxKey = "cc-agent"
// AgentFromCtx returns the authenticated agent name, or "" for console users.
func AgentFromCtx(ctx context.Context) string {
v, _ := ctx.Value(ctxAgent).(string)
return v
}
type Server struct {
cfg config.Config
store *store.Store
agents *AgentAuthenticator
mux *http.ServeMux
}
func New(cfg config.Config, st *store.Store) *Server {
s := &Server{
cfg: cfg,
store: st,
agents: &AgentAuthenticator{st: st},
mux: http.NewServeMux(),
}
s.routes()
return s
}
func (s *Server) Handler() http.Handler { return s.authMiddleware(s.mux) }
func (s *Server) routes() {
s.mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
})
// API route groups are registered by feature packages via Mount.
}
// Mount registers extra routes; used as feature packages (tasks, docs, audit)
// come online.
func (s *Server) Mount(pattern string, h http.HandlerFunc) {
s.mux.HandleFunc(pattern, h)
}
func (s *Server) Store() *store.Store { return s.store }
func (s *Server) Config() config.Config { return s.cfg }
func (s *Server) authMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
p := r.URL.Path
if p == "/healthz" || !strings.HasPrefix(p, "/api/") {
next.ServeHTTP(w, r)
return
}
token := bearerToken(r)
if token == "" {
writeError(w, http.StatusUnauthorized, "missing bearer token")
return
}
if strings.HasPrefix(p, "/api/v1/agent/") {
name, ok := s.agents.Authenticate(r.Context(), token)
if !ok {
writeError(w, http.StatusUnauthorized, "invalid agent token")
return
}
next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), ctxAgent, name)))
return
}
if subtle.ConstantTimeCompare([]byte(token), []byte(s.cfg.ConsoleToken)) != 1 {
writeError(w, http.StatusUnauthorized, "invalid console token")
return
}
next.ServeHTTP(w, r)
})
}
func bearerToken(r *http.Request) string {
h := r.Header.Get("Authorization")
if rest, ok := strings.CutPrefix(h, "Bearer "); ok {
return strings.TrimSpace(rest)
}
if q := r.URL.Query().Get("token"); q != "" && !strings.HasPrefix(r.URL.Path, "/api/") {
return q
}
return ""
}
func TokenHash(token string) string {
sum := sha256.Sum256([]byte(token))
return hex.EncodeToString(sum[:])
}
func writeJSON(w http.ResponseWriter, code int, v any) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(code)
if err := json.NewEncoder(w).Encode(v); err != nil {
slog.Error("write json", "err", err)
}
}
func writeError(w http.ResponseWriter, code int, msg string) {
writeJSON(w, code, map[string]string{"error": msg})
}
+58
View File
@@ -0,0 +1,58 @@
// Package store owns the Postgres connection pool and migrations for the
// Control Center database.
package store
import (
"context"
"database/sql"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/jackc/pgx/v5/stdlib"
"github.com/pressly/goose/v3"
)
type Store struct {
Pool *pgxpool.Pool
}
func Connect(ctx context.Context, dsn string) (*Store, error) {
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
return nil, fmt.Errorf("connect control-center db: %w", err)
}
if err := pool.Ping(ctx); err != nil {
pool.Close()
return nil, fmt.Errorf("ping control-center db: %w", err)
}
return &Store{Pool: pool}, nil
}
func (s *Store) Close() { s.Pool.Close() }
// Migrate applies goose migrations from dir.
func Migrate(ctx context.Context, dsn, dir string) error {
db := stdlib.OpenDBFromPool(mustPool(ctx, dsn))
defer db.Close()
return MigrateDB(db, dir)
}
// MigrateDB applies goose migrations on an existing *sql.DB.
func MigrateDB(db *sql.DB, dir string) error {
goose.SetBaseFS(nil)
if err := goose.SetDialect("postgres"); err != nil {
return err
}
if err := goose.Up(db, dir); err != nil {
return fmt.Errorf("goose up %s: %w", dir, err)
}
return nil
}
func mustPool(ctx context.Context, dsn string) *pgxpool.Pool {
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
panic(fmt.Errorf("connect for migrate: %w", err))
}
return pool
}
+30
View File
@@ -0,0 +1,30 @@
#!/usr/bin/env bash
# Thin wrapper for the personal Control Center. Independent of the product
# system: own DB container (deploy/control-center), own port (8790).
set -euo pipefail
cd "$(dirname "$0")/.."
cmd="${1:-help}"
shift || true
case "$cmd" in
db-start)
docker compose -f deploy/control-center/docker-compose.yml up -d db
;;
db-stop)
docker compose -f deploy/control-center/docker-compose.yml down
;;
serve)
exec go run ./cmd/control-center serve "$@"
;;
migrate)
exec go run ./cmd/control-center migrate "$@"
;;
register-agent)
exec go run ./cmd/control-center register-agent "$@"
;;
*)
echo "usage: $0 {db-start|db-stop|serve|migrate|register-agent} [args...]" >&2
exit 2
;;
esac