diff --git a/cmd/control-center/main.go b/cmd/control-center/main.go new file mode 100644 index 00000000..56292793 --- /dev/null +++ b/cmd/control-center/main.go @@ -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) +} diff --git a/deploy/control-center/docker-compose.yml b/deploy/control-center/docker-compose.yml new file mode 100644 index 00000000..e7a9e24f --- /dev/null +++ b/deploy/control-center/docker-compose.yml @@ -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 diff --git a/deploy/control-center/goose/00001_init.sql b/deploy/control-center/goose/00001_init.sql new file mode 100644 index 00000000..3ff52e5d --- /dev/null +++ b/deploy/control-center/goose/00001_init.sql @@ -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; diff --git a/internal/controlcenter/config/config.go b/internal/controlcenter/config/config.go new file mode 100644 index 00000000..deadaf95 --- /dev/null +++ b/internal/controlcenter/config/config.go @@ -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 /.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 +} diff --git a/internal/controlcenter/server/auth.go b/internal/controlcenter/server/auth.go new file mode 100644 index 00000000..275467c4 --- /dev/null +++ b/internal/controlcenter/server/auth.go @@ -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 +} diff --git a/internal/controlcenter/server/server.go b/internal/controlcenter/server/server.go new file mode 100644 index 00000000..88870d1c --- /dev/null +++ b/internal/controlcenter/server/server.go @@ -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}) +} diff --git a/internal/controlcenter/store/store.go b/internal/controlcenter/store/store.go new file mode 100644 index 00000000..32a67995 --- /dev/null +++ b/internal/controlcenter/store/store.go @@ -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 +} diff --git a/scripts/control-center.sh b/scripts/control-center.sh new file mode 100755 index 00000000..ec95b2cd --- /dev/null +++ b/scripts/control-center.sh @@ -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