diff --git a/cmd/control/main.go b/cmd/control/main.go index 2f4a1266..fa0b27ed 100644 --- a/cmd/control/main.go +++ b/cmd/control/main.go @@ -84,6 +84,13 @@ type Config struct { ValkeyPassword string ValkeyDB int + // rc11 #77: derived-projection reconciler for system actions. + // Interval is the period between full SyncAllUsersSystemActions + // sweeps (safety net for the post-commit listener); 0 disables + // the periodic goroutine entirely. Timeout is the per-sweep + // context deadline so a hung query can't pile up missed ticks. + SystemActionReconcileInterval time.Duration + SystemActionReconcileTimeout time.Duration } func main() { @@ -345,11 +352,46 @@ func main() { logger.Error("failed to reconcile system roles", "error", err) } - // Sync system actions for all users at startup (idempotent) + // rc11 #77: derived-projection wiring for system actions. + // + // 1) One-shot startup sweep — guarantees idempotent convergence + // on every boot, deploy, or upgrade. Logged at Info because it + // runs once. + // 2) Post-commit event listener — fires SyncUserSystemActions + // (or SyncAllUsersSystemActions for fan-out events) on every + // permission-shaping event, so handler tests don't need to + // know about system actions. + // 3) Periodic reconciler — durability safety net for the listener, + // catches any event whose effect on system actions the + // listener doesn't yet know about. Default 1m. if svc.SystemActions() != nil { + // (1) Startup sweep — keeps the existing Info line so + // operators see the one-shot convergence in boot logs. if err := svc.SystemActions().SyncAllUsersSystemActions(ctx); err != nil { logger.Error("failed to sync system actions at startup", "error", err) + } else { + logger.Info("system actions synced for all users (startup)") } + + // (2) Listener — registered post-commit on the store. Logged + // errors are swallowed; the periodic reconciler is the + // durability safety net. Reuse the same per-sweep timeout + // as the reconciler so a wedged DB / signer can't leak a + // goroutine indefinitely (#77 review round 2). + st.RegisterEventListener(api.SystemActionListener( + svc.SystemActions(), + logger.With("component", "system_action_listener"), + cfg.SystemActionReconcileTimeout, + )) + + // (3) Periodic reconciler — interval and per-sweep timeout + // from config (defaults set in parseFlags). + svc.SystemActions().StartReconciliation(ctx, + cfg.SystemActionReconcileInterval, + cfg.SystemActionReconcileTimeout) + logger.Info("system-action reconciliation started", + "interval", cfg.SystemActionReconcileInterval, + "sweep_timeout", cfg.SystemActionReconcileTimeout) } // Configure trusted proxies for X-Forwarded-For header validation if len(cfg.TrustedProxies) > 0 { @@ -470,19 +512,42 @@ func main() { // Index audit events on insertion — the hook fires after every AppendEvent // and enqueues the persisted row directly (no DB lookup in the search worker). - st.OnEventAppended = func(ctx context.Context, ev store.PersistedEvent) { + // Registered via RegisterEventListener so it shares the listener-slice + // mutex + panic-recovery wrapper with every other consumer. + // + // The EnqueueReindex call itself is dispatched in a goroutine so a slow + // or unreachable Valkey cannot stall AppendEvent — fireListeners + // dispatches synchronously, so a blocking listener body would extend + // every state-changing RPC's tail latency by the Valkey RTT. The work + // is best-effort (already only logs Warn on failure), so detaching is + // safe; the goroutine has its own recover so a panic inside the + // taskqueue client can't crash the server. Round-5 review fix. + st.RegisterEventListener(func(ctx context.Context, ev store.PersistedEvent) { id := ulid.ULID(ev.ID).String() - if err := searchIdx.EnqueueReindex(ctx, search.ScopeAuditEvent, id, &taskqueue.SearchEntityData{ + data := &taskqueue.SearchEntityData{ EventType: ev.EventType, StreamType: ev.StreamType, ActorType: ev.ActorType, ActorID: ev.ActorID, StreamID: ev.StreamID, OccurredAt: ev.OccurredAt.Unix(), - }); err != nil { - logger.Warn("failed to enqueue audit event reindex", "id", id, "error", err) } - } + // Detach from the AppendEvent ctx — the RPC may already have + // returned by the time the enqueue runs; cancellation would + // drop best-effort work that the search worker can otherwise + // still pick up. Background ctx is correct here because the + // taskqueue client has its own per-call timeouts. + go func() { + defer func() { + if r := recover(); r != nil { + logger.Error("audit-index listener: panicked", "id", id, "panic", r) + } + }() + if err := searchIdx.EnqueueReindex(context.Background(), search.ScopeAuditEvent, id, data); err != nil { + logger.Warn("failed to enqueue audit event reindex", "id", id, "error", err) + } + }() + }) // Ensure indexes exist (idempotent, needed for FT.SEARCH queries). if err := searchIdx.EnsureIndexes(ctx); err != nil { @@ -584,8 +649,10 @@ func main() { path, handler := pmv1connect.NewControlServiceHandler(svc, interceptors) mux.Handle(path, handler) - // Mount SCIM v2 handler - scimHandler := scim.NewHandler(st, logger) + // Mount SCIM v2 handler. Passes svc.SystemActions() so the SCIM + // delete path can clean up pm-tty-* / USER actions when the + // last identity link is removed (rc11 #77). + scimHandler := scim.NewHandler(st, logger, svc.SystemActions()) mux.Handle("/scim/v2/", scimHandler) // Wrap with CORS and security headers middleware @@ -732,6 +799,11 @@ func parseFlags() *Config { flag.StringVar(&cfg.GatewayURL, "gateway-url", "", "Gateway URL returned to agents during registration") flag.StringVar(&cfg.TerminalGatewayURL, "terminal-gateway-url", "", "Public WebSocket URL of the gateway terminal endpoint (e.g. wss://gw.example.com/terminal). When empty, ControlService.StartTerminal returns CodeUnavailable.") flag.DurationVar(&cfg.DynamicGroupEvalInterval, "dynamic-group-eval-interval", time.Hour, "Interval for evaluating dynamic groups (min 30m, max 8h, 0 to disable)") + // rc11 #77 — system-action reconciliation defaults: 1m interval keeps drift bounded for an + // operator-visible UX path (role grant → terminal works), 5m sweep timeout is plenty for + // even a 10k-user fleet because the sync is read-heavy and short-circuits on no-op users. + flag.DurationVar(&cfg.SystemActionReconcileInterval, "system-action-reconcile-interval", time.Minute, "Period between full SyncAllUsersSystemActions sweeps; 0 disables periodic reconciliation") + flag.DurationVar(&cfg.SystemActionReconcileTimeout, "system-action-reconcile-timeout", 5*time.Minute, "Per-sweep context deadline for the periodic reconciler") flag.StringVar(&cfg.CATrustBundlePath, "ca-trust-bundle", "", "PEM file with trusted CA certificates for verification (supports CA rotation)") flag.BoolVar(&cfg.TLSEnabled, "tls", false, "Enable TLS on public listener") flag.StringVar(&cfg.TLSCert, "tls-cert", "", "TLS certificate for public listener") @@ -763,6 +835,8 @@ func parseFlags() *Config { envString(&cfg.TerminalGatewayURL, "CONTROL_TERMINAL_GATEWAY_URL") envCSV(&cfg.CORSOrigins, "CONTROL_CORS_ORIGINS") envDuration(&cfg.DynamicGroupEvalInterval, "CONTROL_DYNAMIC_GROUP_EVAL_INTERVAL") + envDuration(&cfg.SystemActionReconcileInterval, "CONTROL_SYSTEM_ACTION_RECONCILE_INTERVAL") + envDuration(&cfg.SystemActionReconcileTimeout, "CONTROL_SYSTEM_ACTION_RECONCILE_TIMEOUT") // SSO / Identity Provider configuration cfg.PasswordAuthEnabled = true // default enabled @@ -789,6 +863,27 @@ func parseFlags() *Config { } } + // Clamp system-action reconcile flags. Mirrors the DynamicGroup + // pattern above. A 0 sweep timeout would make + // context.WithTimeout return an already-cancelled context every + // tick, silently breaking the durability safety net; a negative + // interval would panic time.NewTicker. Round-3 review of rc11 + // #77 caught the timeout footgun specifically; round-5 review + // added the floor/ceiling on the interval so a misconfigured + // 1ms tick can't flood the DB with sweep attempts. + if cfg.SystemActionReconcileInterval < 0 { + cfg.SystemActionReconcileInterval = 0 // treat as disabled, matching StartReconciliation + } else if cfg.SystemActionReconcileInterval > 0 { + if cfg.SystemActionReconcileInterval < 10*time.Second { + cfg.SystemActionReconcileInterval = 10 * time.Second + } else if cfg.SystemActionReconcileInterval > 8*time.Hour { + cfg.SystemActionReconcileInterval = 8 * time.Hour + } + } + if cfg.SystemActionReconcileTimeout <= 0 { + cfg.SystemActionReconcileTimeout = 5 * time.Minute + } + if cfg.JWTSecret == "" { fmt.Fprintln(os.Stderr, "FATAL: CONTROL_JWT_SECRET (or -jwt-secret) is required") os.Exit(1) diff --git a/cmd/gateway/main.go b/cmd/gateway/main.go index 9a92cb65..4cc93910 100644 --- a/cmd/gateway/main.go +++ b/cmd/gateway/main.go @@ -56,10 +56,15 @@ func main() { slog.SetDefault(logger) // Config-shape checks that don't fit the simple "required env - // var empty" pattern below (TTY/MTLS host collision, etc.). - // Failing here keeps them visible at startup rather than at the - // first affected request. - if err := cfg.Validate(); err != nil { + // var empty" pattern below (TTY/MTLS host collision, partial + // terminal config, etc.). Failing here keeps fatal issues + // visible at startup; warnings surface partial misconfigurations + // that would otherwise produce silent runtime failures. + warnings, err := cfg.Validate() + for _, w := range warnings { + logger.Warn("gateway configuration warning", "warning", w) + } + if err != nil { logger.Error("invalid gateway configuration", "error", err) os.Exit(1) } diff --git a/deploy/.env.example b/deploy/.env.example index bbd8eb7c..3d2f39c6 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -149,6 +149,22 @@ LOG_LEVEL=info # Min: 30m, Max: 8h, 0 to disable. Default: 1h # DYNAMIC_GROUP_EVAL_INTERVAL=1h +# --- System Action Reconciliation (rc11 #77) -------------------------------- +# Period between full system-action reconciliation sweeps. The control +# server runs an event-driven listener for sub-second convergence after +# permission-shaping mutations; this periodic sweep is the durability +# safety net for any event the listener doesn't yet classify, and for +# the rare crash-between-commit-and-listener case. 1m default keeps +# drift bounded for the operator-visible "grant role → terminal works" +# UX. Set to 0 to disable the safety net entirely (NOT recommended). +# CONTROL_SYSTEM_ACTION_RECONCILE_INTERVAL=1m + +# Per-sweep context deadline. A hung query gets cancelled after this +# rather than piling up missed ticks behind it. 5m is plenty for a +# 10k-user fleet because SyncAllUsersSystemActions is read-heavy and +# short-circuits on no-op users. +# CONTROL_SYSTEM_ACTION_RECONCILE_TIMEOUT=5m + # --- Network ---------------------------------------------------------------- # Comma-separated trusted proxy IPs/CIDRs for correct client IP parsing. # Set this when running behind a reverse proxy (Traefik is added automatically). diff --git a/deploy/QUICKSTART.md b/deploy/QUICKSTART.md new file mode 100644 index 00000000..7a26a0ba --- /dev/null +++ b/deploy/QUICKSTART.md @@ -0,0 +1,81 @@ +# Power Manage Server — Quickstart + +One-line install on a fresh Linux host with Docker + the compose plugin already installed: + +```bash +curl -fsSL https://raw.githubusercontent.com/manchtools/power-manage-server/main/deploy/install.sh | sudo bash +``` + +The installer: +1. Verifies `docker` + `docker compose` are present (does **not** install them — you own that step). +2. Downloads the `deploy/` tree from the latest pre-release tag. +3. Copies `.env.example` to `.env` if missing. +4. Runs `setup.sh` in **guided mode** — interactive prompts for domains, ACME email, and admin credentials; auto-generates strong defaults for every secret (Postgres / indexer / Valkey passwords, JWT secret, encryption key). +5. Pulls the Power Manage container images. +6. Brings the stack up via `docker compose up -d`. +7. Prints the URLs and next steps. + +## Pinning a specific version + +The installer defaults to `latest-rc` (the curated pre-release tag). For a stable release or a specific RC: + +```bash +curl -fsSL .../install.sh | sudo RELEASE_TAG=v2026.05 bash +``` + +> **Note:** the env var goes between `sudo` and `bash`, **not** before `curl`. By default `sudo` resets the environment (`env_reset`), so `RELEASE_TAG=… curl … | sudo bash` would only set the variable for the local `curl` process and the installer running under `sudo` would never see it. Putting it after `sudo` passes it through. + +## Non-interactive install + +CI / Ansible / preconfigured `.env` setups can skip the guided prompts: + +```bash +curl -fsSL .../install.sh | sudo NO_PROMPT=1 bash +``` + +The installer expects `.env` (or `.env.example` to copy from) at `INSTALL_DIR` and runs `setup.sh --no-prompt` — strict env validation only, no prompts. + +## Custom install directory + +```bash +curl -fsSL .../install.sh | sudo INSTALL_DIR=/srv/pm bash +``` + +## Re-running the installer + +`install.sh` is idempotent. Re-running on an existing install: +- Preserves `.env` (your secrets stay intact). +- Pulls the chosen `RELEASE_TAG` images. +- Restarts the stack. + +Use this for upgrades: + +```bash +sudo RELEASE_TAG=v2026.06 bash /opt/power-manage/install.sh # if previously installed +# or fetch a fresh installer +curl -fsSL .../install.sh | sudo RELEASE_TAG=v2026.06 bash +``` + +## Manual install (if you'd rather not run a curl-pipe-bash) + +```bash +git clone https://github.com/manchtools/power-manage-server.git +cd power-manage-server/deploy +cp .env.example .env +./setup.sh # guided env + cert generation +docker compose up -d +``` + +## What the installer does NOT do + +- **Install Docker.** Use [the official convenience script](https://docs.docker.com/engine/install/) or your distro's package manager first. +- **Configure DNS.** You need A/AAAA records pointing your `CONTROL_DOMAIN`, `GATEWAY_DOMAIN`, and (if terminals are enabled) `GATEWAY_TTY_DOMAIN` at this host. +- **Open firewall ports.** Traefik binds `:80` (LE http-01 challenge + redirect-to-https) and `:443` (everything else) on the host. Open those before the first start so Let's Encrypt can issue certificates. +- **Migrate from a pre-rc11 deploy.** This installer is for fresh installs and same-release upgrades. Migrating across breaking releases follows the per-release migration runbook. + +## After install + +1. Wait ~30 s for Let's Encrypt to issue certs on first run. +2. Log in to `https://` with the bootstrap admin credentials the installer printed. +3. Create real user accounts (UI, SSO, or SCIM) — the bootstrap admin is intentionally not for daily use; see [`.env.example`](./.env.example) for details. +4. Generate a registration token and enroll your first agent. diff --git a/deploy/install.sh b/deploy/install.sh new file mode 100755 index 00000000..0f3e1572 --- /dev/null +++ b/deploy/install.sh @@ -0,0 +1,276 @@ +#!/usr/bin/env bash +# +# Power Manage Server bootstrap installer. +# +# curl -fsSL https://raw.githubusercontent.com/MANCHTOOLS/power-manage-server/main/deploy/install.sh | sudo bash +# +# Pulls the deploy/ tree from a chosen release tag, runs setup.sh +# (interactive guided mode by default), pulls images, brings the +# stack up. Does NOT install Docker — checks for it and fails with a +# clear message if missing. Operator owns Docker setup via their +# preferred channel (apt / dnf / official convenience script). +# +# Idempotent: re-running on an existing install does not clobber +# .env or regenerate certs; offers to update IMAGE_TAG and restart. +# +# Env-var overrides: +# INSTALL_DIR Target directory (default: /opt/power-manage) +# RELEASE_TAG Image + deploy tag (default: latest-rc) +# NO_PROMPT Set to 1 to skip the guided setup loop +# +# rc11 #80. + +set -euo pipefail + +INSTALL_DIR="${INSTALL_DIR:-/opt/power-manage}" +RELEASE_TAG="${RELEASE_TAG:-latest-rc}" +NO_PROMPT="${NO_PROMPT:-0}" +GITHUB_REPO="manchtools/power-manage-server" + +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +NC='\033[0m' + +log_info() { echo -e "${GREEN}[INFO]${NC} $1"; } +log_warn() { echo -e "${YELLOW}[WARN]${NC} $1"; } +log_error() { echo -e "${RED}[ERROR]${NC} $1" >&2; } + +############################################################################### +# Step 1 — preflight +# +# We DO NOT install Docker. Too many distro/version permutations to +# do safely from a generic installer; operators handle that via the +# official Docker convenience script or their package manager. We +# just check that both pieces are present and runnable as the +# current user. +############################################################################### +preflight() { + log_info "Preflight checks…" + + if ! command -v docker >/dev/null 2>&1; then + log_error "docker not found. Install Docker first via your distro's package manager" + log_error " or the official convenience script: https://docs.docker.com/engine/install/" + exit 1 + fi + + if ! docker compose version >/dev/null 2>&1; then + log_error "docker compose plugin not found. Install via:" + log_error " Debian/Ubuntu: sudo apt install docker-compose-plugin" + log_error " Fedora: sudo dnf install docker-compose-plugin" + log_error " Or per-distro instructions: https://docs.docker.com/compose/install/" + exit 1 + fi + + if ! docker info >/dev/null 2>&1; then + log_error "docker daemon not reachable. Either start it (sudo systemctl start docker)" + log_error " or run install.sh as a user in the docker group / as root." + exit 1 + fi + + if ! command -v curl >/dev/null 2>&1; then + log_error "curl not found — required for downloading the release tarball." + exit 1 + fi + + if ! command -v tar >/dev/null 2>&1; then + log_error "tar not found — required for extracting the release tarball." + exit 1 + fi + + log_info " docker: $(docker --version | cut -d, -f1)" + log_info " docker compose: $(docker compose version --short 2>/dev/null || echo unknown)" + log_info " install dir: $INSTALL_DIR" + log_info " release tag: $RELEASE_TAG" +} + +############################################################################### +# Step 2 — fetch the deploy/ tree +# +# Downloads the release tarball from GitHub and extracts only the +# server/deploy/ subdirectory into INSTALL_DIR. Avoids needing git +# on the host for what is fundamentally a deploy-artifact pull. +# +# RELEASE_TAG can be a specific tag (v2026.05-rc11) OR a curated +# floating tag (latest-rc, main). We resolve via the GitHub +# tarball-download endpoint which accepts both. +############################################################################### +download_deploy_tree() { + log_info "Fetching deploy tree from $GITHUB_REPO@$RELEASE_TAG…" + + local tmpdir + tmpdir="$(mktemp -d)" + # Single-quote so $tmpdir is expanded at trap-fire time, not now — + # SC2064. Works either way today (the var isn't reassigned), but + # the deferred-expansion form is the documented intent. + trap 'rm -rf "$tmpdir"' EXIT + + local tarball="$tmpdir/source.tar.gz" + + # GitHub serves tarballs at /{owner}/{repo}/archive/refs/{heads,tags}/{ref}.tar.gz. + # We try tags first (release case), then heads (branch case for "main"). + local url_tag="https://github.com/${GITHUB_REPO}/archive/refs/tags/${RELEASE_TAG}.tar.gz" + local url_branch="https://github.com/${GITHUB_REPO}/archive/refs/heads/${RELEASE_TAG}.tar.gz" + + if curl -fsSL "$url_tag" -o "$tarball" 2>/dev/null; then + log_info " Resolved as tag: $RELEASE_TAG" + elif curl -fsSL "$url_branch" -o "$tarball" 2>/dev/null; then + log_info " Resolved as branch: $RELEASE_TAG" + else + log_error "Could not resolve $RELEASE_TAG as either a tag or branch on $GITHUB_REPO." + log_error " Check the spelling, or set RELEASE_TAG to a known value:" + log_error " RELEASE_TAG=latest-rc ./install.sh # latest pre-release (curated)" + log_error " RELEASE_TAG=v2026.05 ./install.sh # specific stable release" + exit 1 + fi + + # Extract only the deploy/ subtree. The tarball top-level + # directory is repo-{tag}, so we strip the first component AND + # restrict to deploy/* paths. + mkdir -p "$INSTALL_DIR" + + if [[ -f "$INSTALL_DIR/.env" ]]; then + log_warn "Existing .env found at $INSTALL_DIR/.env — preserving it (idempotent re-run)." + # Stash it so the extract can overwrite docker-compose.yml / + # setup.sh / etc. without clobbering the operator's secrets. + cp "$INSTALL_DIR/.env" "$tmpdir/.env.preserved" + fi + + # --strip-components=2 turns "repo-tag/deploy/file" into "file". + # The "*/deploy" wildcard restricts extraction to just the + # deploy/ subtree. + tar -xzf "$tarball" \ + -C "$INSTALL_DIR" \ + --strip-components=2 \ + --wildcards \ + '*/deploy/*' + + # tar exits 0 even when the wildcard matches nothing — e.g. a + # future tag that renames deploy/ or a private fork with a + # different layout. Fail loudly here instead of letting the + # operator chase a confusing "setup.sh: command not found" later. + if [[ ! -f "$INSTALL_DIR/setup.sh" ]] || [[ ! -f "$INSTALL_DIR/.env.example" ]]; then + log_error "Extracted tarball but expected files (setup.sh / .env.example) are missing under $INSTALL_DIR." + log_error " The tarball at $RELEASE_TAG may not contain a deploy/ subtree in the expected layout." + exit 1 + fi + + if [[ -f "$tmpdir/.env.preserved" ]]; then + cp "$tmpdir/.env.preserved" "$INSTALL_DIR/.env" + log_info " Preserved .env restored." + fi + + log_info " Extracted to $INSTALL_DIR/" +} + +############################################################################### +# Step 3 — initialise .env if missing +############################################################################### +init_env() { + cd "$INSTALL_DIR" + if [[ ! -f .env ]]; then + cp .env.example .env + log_info "Created .env from .env.example — guided setup will fill it in." + else + log_info ".env already exists — guided setup will only prompt for missing values." + fi +} + +############################################################################### +# Step 4 — run setup.sh (cert generation + guided env loop) +############################################################################### +run_setup() { + cd "$INSTALL_DIR" + if [[ ! -x ./setup.sh ]]; then + chmod +x ./setup.sh + fi + + if [[ "$NO_PROMPT" == "1" ]]; then + ./setup.sh --no-prompt + elif [[ -r /dev/tty ]]; then + # When this installer was piped (the documented + # `curl … | sudo bash` flow), our stdin is the pipe and + # setup.sh's `-t 0` check would auto-fall-back to + # non-interactive mode — which then fails immediately on the + # CHANGE_ME placeholders in a fresh .env. Reattach to the + # controlling terminal so the prompt loop actually runs. + # If /dev/tty is unreadable (true non-interactive context like + # CI without NO_PROMPT=1), the existing -t 0 fallback in + # setup.sh still produces a clean "validate .env directly" + # path with the right error. + ./setup.sh }" + echo " Gateway mTLS: https://${GATEWAY_DOMAIN:-}" + if [[ -n "${GATEWAY_TTY_DOMAIN:-}" ]]; then + echo " TTY traffic: https://${GATEWAY_TTY_DOMAIN}" + fi + echo "" + echo " Admin login: ${ADMIN_EMAIL:-}" + echo " Install dir: $INSTALL_DIR" + echo "" + echo "Next steps:" + echo " 1. Wait ~30s for Let's Encrypt to issue certs (first run only)." + echo " 2. Log in to the Control UI and create real user accounts." + echo " 3. Generate a registration token, then enroll an agent on a device:" + echo " curl -fsSL https://github.com/MANCHTOOLS/power-manage-agent/releases/latest/download/install.sh | sudo bash -s -- -s https://${CONTROL_DOMAIN:-} -t " + echo "" + echo " Logs: docker compose -f $INSTALL_DIR/compose.yml logs -f" + echo "" +} + +main() { + log_info "Power Manage Server installer (rc11)" + echo "" + + preflight + download_deploy_tree + init_env + run_setup + start_stack + print_summary +} + +main "$@" diff --git a/deploy/setup.sh b/deploy/setup.sh index c7a4b3c0..e80d8a28 100755 --- a/deploy/setup.sh +++ b/deploy/setup.sh @@ -280,10 +280,370 @@ show_instructions() { echo "" } +############################################################################### +# Guided env setup (rc11 #80) +# +# Interactive prompt loop that fills in missing .env values for +# operators who'd rather click than read .env.example. Skipped when: +# * --no-prompt is passed +# * stdin is not a TTY (CI, piped input) +# * .env already has every required value (idempotent re-run) +# +# Each prompt: +# * Skips if the variable already has a non-placeholder value +# * Offers to auto-generate strong defaults for secrets +# * Validates URL-safety / hex / hostname constraints inline so the +# operator can fix typos before they cause obscure runtime errors +# * Auto-composes URL-template strings (GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE) +# from the chosen TTY domain — operator never types {id} by hand +############################################################################### + +# is_placeholder returns 0 (truthy) if the value is empty, the +# CHANGE_ME sentinel, or one of the example.com defaults from +# .env.example. Used to decide whether a prompt should fire. +is_placeholder() { + local v="$1" + [[ -z "$v" ]] && return 0 + [[ "$v" == CHANGE_ME* ]] && return 0 + [[ "$v" == *"example.com"* ]] && return 0 + return 1 +} + +# parent_domain returns the part of $1 after the first dot, or empty +# when $1 has no dot at all. Used to derive sibling-subdomain defaults +# without falling into the `${var#*.}` footgun where a no-match returns +# the WHOLE string — that would make `tty.${CONTROL_DOMAIN#*.}` for a +# single-label `localhost` resolve to `tty.localhost`, which is +# self-referential and not a useful prompt default. Round-4 review +# follow-up. +parent_domain() { + local d="$1" + if [[ "$d" == *.* ]]; then + echo "${d#*.}" + else + echo "" + fi +} + +# write_env_var atomically updates a single key=value line in .env. +# Adds the key if missing; preserves surrounding comments/order. +# +# Atomicity: the temp file is created in the same directory as .env so +# `mv` resolves to a same-filesystem rename(2), which is atomic on +# POSIX. Default mktemp uses $TMPDIR (often a separate mount), in +# which case mv falls back to copy-then-unlink — non-atomic and also +# overwrites .env's mode/owner with the temp file's. Mode is copied +# explicitly with chmod --reference so re-runs don't downgrade .env +# off 0600. +write_env_var() { + local key="$1" value="$2" envfile="$SCRIPT_DIR/.env" + if grep -qE "^${key}=" "$envfile"; then + local tmp + tmp="$(mktemp "${envfile}.XXXXXX")" + awk -v k="$key" -v v="$value" ' + BEGIN { found = 0 } + $0 ~ "^"k"=" { print k"="v; found = 1; next } + { print } + END { if (!found) print k"="v } + ' "$envfile" > "$tmp" + chmod --reference="$envfile" "$tmp" 2>/dev/null || chmod 600 "$tmp" + mv "$tmp" "$envfile" + else + printf '%s=%s\n' "$key" "$value" >> "$envfile" + fi +} + +# clear_env_var removes a key=value line from .env entirely. Used by +# the guided setup when the operator answers No to a feature on a +# rerun where existing values would otherwise leave the feature +# silently enabled (caught in #80 review). No-op when the key is +# absent. Same-filesystem mktemp + mode preservation as write_env_var. +clear_env_var() { + local key="$1" envfile="$SCRIPT_DIR/.env" + if ! grep -qE "^${key}=" "$envfile"; then + return 0 + fi + local tmp + tmp="$(mktemp "${envfile}.XXXXXX")" + awk -v k="$key" ' + $0 ~ "^"k"=" { next } + { print } + ' "$envfile" > "$tmp" + chmod --reference="$envfile" "$tmp" 2>/dev/null || chmod 600 "$tmp" + mv "$tmp" "$envfile" +} + +# prompt_secret asks for a secret value, offers to generate one with +# the supplied openssl command. Stores the chosen value in $REPLY_VALUE +# and a "did the operator just choose a value?" flag in $REPLY_GENERATED +# (1 = newly generated/entered this run, 0 = kept existing). Callers +# use the flag to decide whether to print the value back as a one-time +# capture banner — re-printing on every rerun would leak the stored +# password to terminal scrollback every time setup.sh is re-run. +# Round-5 review fix. +# +# The manual-entry branch uses `read -s` so the typed secret is never +# echoed back to the terminal — the auto-generate path never traverses +# stdin so it's already silent. +prompt_secret() { + local prompt="$1" gen_cmd="$2" current="$3" + REPLY_VALUE="" + REPLY_GENERATED=0 + if ! is_placeholder "$current"; then + log_info " $prompt — already set, keeping current value" + REPLY_VALUE="$current" + return 0 + fi + echo "" + read -r -p " $prompt — generate strong value? [Y/n] " ans + if [[ -z "$ans" || "$ans" =~ ^[Yy] ]]; then + REPLY_VALUE="$(eval "$gen_cmd")" + REPLY_GENERATED=1 + echo " ✓ Generated." + else + read -r -s -p " Enter value: " REPLY_VALUE + REPLY_GENERATED=1 + # `read -s` suppresses the trailing newline; print one so the + # subsequent log lines start on a fresh row. + echo + fi +} + +# prompt_string asks for a free-form value, with optional default. +# Stores the chosen value in $REPLY_VALUE. +prompt_string() { + local prompt="$1" default="$2" current="$3" + REPLY_VALUE="" + if ! is_placeholder "$current"; then + log_info " $prompt — already set ($current), keeping" + REPLY_VALUE="$current" + return 0 + fi + echo "" + if [[ -n "$default" ]]; then + read -r -p " $prompt [$default]: " REPLY_VALUE + [[ -z "$REPLY_VALUE" ]] && REPLY_VALUE="$default" + else + read -r -p " $prompt: " REPLY_VALUE + fi +} + +# prompt_yes_no asks a Y/n question, defaults to Yes. +prompt_yes_no() { + local prompt="$1" default_yes="${2:-yes}" + local hint="[Y/n]" + [[ "$default_yes" != "yes" ]] && hint="[y/N]" + echo "" + read -r -p " $prompt $hint " ans + if [[ -z "$ans" ]]; then + [[ "$default_yes" == "yes" ]] && return 0 || return 1 + fi + [[ "$ans" =~ ^[Yy] ]] +} + +# guided_setup runs the interactive prompt loop. Invoked from main() +# only when stdin is a TTY and --no-prompt was not passed. +guided_setup() { + log_info "Guided setup — prompting for missing values." + echo " Press Ctrl-C at any time to abort. Existing .env values are kept." + echo "" + + # Source current values so prompts can detect "already set". + set -a + [[ -f "$SCRIPT_DIR/.env" ]] && source "$SCRIPT_DIR/.env" + set +a + + # --- Domains --- + prompt_string "Control server public domain (CONTROL_DOMAIN)" "" "${CONTROL_DOMAIN:-}" + write_env_var CONTROL_DOMAIN "$REPLY_VALUE" + CONTROL_DOMAIN="$REPLY_VALUE" + + prompt_string "Gateway domain — agent mTLS endpoint (GATEWAY_DOMAIN)" "" "${GATEWAY_DOMAIN:-}" + write_env_var GATEWAY_DOMAIN "$REPLY_VALUE" + GATEWAY_DOMAIN="$REPLY_VALUE" + + # Terminal sessions are optional but recommended; offer the full set. + if prompt_yes_no "Enable remote terminal (TTY) sessions?"; then + # Validate distinct host inline so the rc10 collision check + # never fires. + local default_tty="" + local control_parent + control_parent="$(parent_domain "$CONTROL_DOMAIN")" + if [[ -n "$control_parent" ]]; then + default_tty="tty.$control_parent" + fi + # Single-label CONTROL_DOMAIN (e.g. `localhost`) leaves the + # default empty so the operator types something meaningful + # rather than accepting `tty.localhost`. + prompt_string "TTY domain (must differ from GATEWAY_DOMAIN)" "$default_tty" "${GATEWAY_TTY_DOMAIN:-}" + if [[ "$REPLY_VALUE" == "$GATEWAY_DOMAIN" ]]; then + log_error "GATEWAY_TTY_DOMAIN must differ from GATEWAY_DOMAIN; aborting" + log_error " Traefik TCP-passthrough for mTLS would shadow the TTY HTTP router on a shared SNI." + exit 1 + fi + write_env_var GATEWAY_TTY_DOMAIN "$REPLY_VALUE" + local tty_dom="$REPLY_VALUE" + + # Auto-compose the URL template. Operator never types {id}. + write_env_var GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE "wss://${tty_dom}/gw/{id}/terminal" + echo " ✓ GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE composed automatically." + + # The TTY HTTP listener inside the container — Traefik + # terminates public TLS and forwards cleartext. + write_env_var GATEWAY_WEB_LISTEN_ADDR ":8443" + echo " ✓ GATEWAY_WEB_LISTEN_ADDR set to :8443." + else + # Operator chose No. If an existing .env already has any of + # these set (e.g. a rerun where terminals were previously + # enabled), simply skipping leaves the feature on — the + # gateway would still publish its terminal URL on next boot. + # Clear all three explicitly so the No answer matches the + # observable state. Caught in #80 review. + local was_enabled=0 + for k in GATEWAY_TTY_DOMAIN GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE GATEWAY_WEB_LISTEN_ADDR; do + if grep -qE "^${k}=" "$SCRIPT_DIR/.env" 2>/dev/null; then + was_enabled=1 + clear_env_var "$k" + fi + done + if [[ "$was_enabled" -eq 1 ]]; then + log_info " Terminal sessions disabled — cleared GATEWAY_TTY_DOMAIN, GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE, and GATEWAY_WEB_LISTEN_ADDR from .env." + else + log_info " Terminal sessions disabled — none of the terminal env vars were set, nothing to clear." + fi + fi + + prompt_string "Email for Let's Encrypt notifications (ACME_EMAIL)" "" "${ACME_EMAIL:-}" + write_env_var ACME_EMAIL "$REPLY_VALUE" + + # --- Image tag --- + prompt_string "Image tag — :latest, :latest-rc, or pin to vYYYY.MM (IMAGE_TAG)" "latest" "${IMAGE_TAG:-}" + write_env_var IMAGE_TAG "$REPLY_VALUE" + + # --- Secrets --- + echo "" + log_info "Generating / collecting secrets…" + + prompt_secret "PostgreSQL password (POSTGRES_PASSWORD)" "openssl rand -base64 32" "${POSTGRES_PASSWORD:-}" + write_env_var POSTGRES_PASSWORD "$REPLY_VALUE" + + # Indexer password MUST be URL-safe — used in a libpq DSN by the + # indexer. Hex output is the safest generator for this constraint. + prompt_secret "Indexer DB password (INDEXER_POSTGRES_PASSWORD, must be URL-safe)" "openssl rand -hex 32" "${INDEXER_POSTGRES_PASSWORD:-}" + if [[ "$REPLY_VALUE" =~ [^A-Za-z0-9_.-] ]]; then + log_error "INDEXER_POSTGRES_PASSWORD contains URL-unsafe characters: ${BASH_REMATCH[0]}" + log_error " Use 'openssl rand -hex 32' or pick alphanumeric only. Aborting." + exit 1 + fi + write_env_var INDEXER_POSTGRES_PASSWORD "$REPLY_VALUE" + + prompt_secret "Valkey password (VALKEY_PASSWORD)" "openssl rand -base64 32" "${VALKEY_PASSWORD:-}" + write_env_var VALKEY_PASSWORD "$REPLY_VALUE" + + prompt_secret "JWT secret (JWT_SECRET, min 32 chars)" "openssl rand -base64 48" "${JWT_SECRET:-}" + if [[ ${#REPLY_VALUE} -lt 32 ]]; then + log_error "JWT_SECRET must be at least 32 characters; got ${#REPLY_VALUE}. Aborting." + exit 1 + fi + write_env_var JWT_SECRET "$REPLY_VALUE" + + # Encryption key must be exactly 64 hex chars (32 bytes). + prompt_secret "Encryption key for IdP/LUKS secrets (CONTROL_ENCRYPTION_KEY, 64 hex chars)" "openssl rand -hex 32" "${CONTROL_ENCRYPTION_KEY:-}" + if [[ ! "$REPLY_VALUE" =~ ^[0-9a-fA-F]{64}$ ]]; then + log_error "CONTROL_ENCRYPTION_KEY must be exactly 64 hex characters; got ${#REPLY_VALUE} chars. Aborting." + exit 1 + fi + write_env_var CONTROL_ENCRYPTION_KEY "$REPLY_VALUE" + + # --- Admin account --- + # admin@ if CONTROL_DOMAIN has a dot; admin@ + # for single-label cases (admin@localhost is a valid local-delivery + # address for those deployments). + local admin_parent + admin_parent="$(parent_domain "$CONTROL_DOMAIN")" + local default_email + if [[ -n "$admin_parent" ]]; then + default_email="admin@$admin_parent" + else + default_email="admin@$CONTROL_DOMAIN" + fi + prompt_string "Bootstrap admin email (ADMIN_EMAIL)" "$default_email" "${ADMIN_EMAIL:-}" + write_env_var ADMIN_EMAIL "$REPLY_VALUE" + # Mirror the prompt result back into the shell var. The summary + # block below would otherwise echo the stale value sourced before + # the prompt loop ran (empty on a fresh install). + ADMIN_EMAIL="$REPLY_VALUE" + + prompt_secret "Bootstrap admin password (ADMIN_PASSWORD)" "openssl rand -base64 24" "${ADMIN_PASSWORD:-}" + write_env_var ADMIN_PASSWORD "$REPLY_VALUE" + local admin_pass="$REPLY_VALUE" + local admin_pass_generated="$REPLY_GENERATED" + + echo "" + log_info "Guided setup complete. .env updated." + # Only print the password back when it was newly chosen this run. + # Reusing is_placeholder here was wrong: a real password is not a + # placeholder, so on every rerun the banner would re-leak the + # stored password into the terminal scrollback / install logs. + # Round-5 review fix. + if [[ "$admin_pass_generated" -eq 1 ]]; then + log_warn "Bootstrap admin password — write this down NOW; it's not shown again:" + echo "" + echo " Email: $ADMIN_EMAIL" + echo " Password: $admin_pass" + echo "" + log_warn "The bootstrap admin is for first-login only — see deploy/.env.example for details." + fi + echo "" +} + +# parse_flags reads our own --no-prompt before falling through to the +# rest of setup.sh. Kept simple — no other flags supported. Wrapped +# in a function so sourcing setup.sh from setup_test.sh doesn't pick +# up the harness's argv (round-5 review: source-guard pattern lets +# the test harness exercise the real helper bodies instead of inlined +# copies that drift). +NO_PROMPT=0 +parse_flags() { + for arg in "$@"; do + case "$arg" in + --no-prompt) NO_PROMPT=1 ;; + -h|--help) + cat < "$SCRIPT_DIR/.env" + write_env_var FOO bar + grep -qE '^FOO=bar$' "$SCRIPT_DIR/.env" +} + +case_write_env_var_updates_existing_key() { + cat > "$SCRIPT_DIR/.env" < "$SCRIPT_DIR/.env" < "$SCRIPT_DIR/.env" < "$SCRIPT_DIR/.env" < "$SCRIPT_DIR/.env" + chmod 600 "$SCRIPT_DIR/.env" + write_env_var FOO bar + write_env_var FOO baz # exercise the rewrite branch specifically + local mode + mode="$(stat -c '%a' "$SCRIPT_DIR/.env")" + [[ "$mode" == "600" ]] +} + +case_clear_env_var_preserves_mode_0600() { + cat > "$SCRIPT_DIR/.env" < "$SCRIPT_DIR/.env" + write_env_var ISOLATION_PROBE yes + [[ "$SCRIPT_DIR" == "$(dirname "$SCRIPT_DIR")"/* ]] || return 1 + [[ -f "$SCRIPT_DIR/.env" ]] || return 1 + grep -qE '^ISOLATION_PROBE=yes$' "$SCRIPT_DIR/.env" +} + +case_parent_domain_with_dot() { + : > "$SCRIPT_DIR/.env" + local got + got="$(parent_domain control.example.com)" + [[ "$got" == "example.com" ]] +} + +case_parent_domain_single_label() { + : > "$SCRIPT_DIR/.env" + local got + got="$(parent_domain localhost)" + [[ -z "$got" ]] +} + +case_parent_domain_deep_subdomain() { + : > "$SCRIPT_DIR/.env" + local got + got="$(parent_domain a.b.c.example.com)" + [[ "$got" == "b.c.example.com" ]] +} + +case_disable_terminals_clears_all_three_vars() { + # Simulates the rerun footgun the review caught: existing .env + # has all three terminal vars set; operator answers No; the + # disable path must clear all three. + cat > "$SCRIPT_DIR/.env" < 0 { m.logger.Warn("some users failed system action sync", "failed", errCount, "total", len(users)) } else { - m.logger.Info("system actions synced for all users", "count", len(users)) + // Demoted from Info to Debug in rc11 (#77): the periodic + // reconciler runs this on a tight cadence (default 1m), so + // success-on-every-tick at Info would flood operator logs. + // Startup callers in cmd/control/main.go log their own Info + // line so the one-shot startup sweep stays visible. + m.logger.Debug("system actions synced for all users", "count", len(users)) } return nil } +// StartReconciliation launches a background goroutine that periodically +// runs SyncAllUsersSystemActions as the durability safety net for the +// event-driven listener (see system_actions_listener.go). Closes drift +// gaps if the listener fires post-commit but the process dies before +// the sync runs, plus any future event type added to the schema +// without being added to the AffectedFromEvent classifier. +// +// Guards: +// - atomic flag prevents a slow sweep from stacking another one +// behind it under DB pressure / large fleets; +// - per-sweep timeout cancels a stuck invocation rather than piling +// up missed ticks; +// - SyncAllUsersSystemActions is already best-effort per user, so +// one bad user cannot abort the sweep. +// +// rc11 #77. +func (m *SystemActionManager) StartReconciliation(ctx context.Context, interval, sweepTimeout time.Duration) { + if interval <= 0 { + m.logger.Info("system-action reconciliation disabled (interval <= 0)") + return + } + // A non-positive sweepTimeout would feed an already-cancelled + // context into SyncAllUsersSystemActions on every tick — the + // reconciler would log an error every interval and never make + // progress. parseFlags also clamps env input, but defend in depth + // here so a buggy programmatic caller can't silently break the + // safety net. Round-3 review of rc11 #77. + if sweepTimeout <= 0 { + m.logger.Warn("system-action reconciliation sweep timeout <= 0; falling back to interval as ceiling", + "sweep_timeout", sweepTimeout, "interval", interval) + sweepTimeout = interval + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + + var running atomic.Bool + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if !running.CompareAndSwap(false, true) { + m.logger.Warn("skipping system-action reconciliation tick — previous sweep still running") + continue + } + sweepCtx, cancel := context.WithTimeout(ctx, sweepTimeout) + if err := m.SyncAllUsersSystemActions(sweepCtx); err != nil { + m.logger.Error("periodic system-action reconciliation failed", "error", err) + } + cancel() + running.Store(false) + } + } + }() +} + // SyncUserSystemActions ensures a user's system actions are up to date. // Actions are only created when provisioning is enabled (globally or per-user). // When provisioning is disabled, existing actions are cleaned up. diff --git a/internal/api/system_actions_listener.go b/internal/api/system_actions_listener.go new file mode 100644 index 00000000..ce84b3f0 --- /dev/null +++ b/internal/api/system_actions_listener.go @@ -0,0 +1,271 @@ +package api + +import ( + "context" + "encoding/json" + "log/slog" + "strings" + "sync/atomic" + "time" + + "github.com/manchtools/power-manage/server/internal/store" +) + +// SyncOp is the action a system-action listener should take in response +// to a permission-shaping event. The "all-users" cases (role / group +// fan-outs, server-settings) deliberately route to a full sweep rather +// than materialising a potentially-large affected-user list in the +// classifier path — the manager's full sweep is already efficient and +// one sweep is cheaper than thousands of per-user enqueues on a big +// fleet. rc11 #77. +type SyncOp int + +const ( + SyncOpNone SyncOp = iota // event does not affect system actions + SyncOpSyncUser // SyncUserSystemActions(userID) + SyncOpCleanupUser // CleanupDeletedUserActions(userProjection) + SyncOpSyncAll // SyncAllUsersSystemActions — fan-out events +) + +// AffectedFromEvent classifies an event into the system-action sync +// operation it should trigger and (for per-user ops) the user IDs +// involved. +// +// Adding a new permission-shaping event in a future handler only +// requires adding a case here. The periodic reconciler is the safety +// net for events the classifier has not yet learned about, so a missed +// case manifests as bounded drift (~one reconcile interval), not +// silent failure. +func AffectedFromEvent(e store.PersistedEvent) (SyncOp, []string) { + switch e.EventType { + // Direct user-scoped events — system actions for the named user + // must be re-evaluated. The user_id source varies: for `user` + // stream events it's the StreamID; for `user_role` it's in the + // event Data payload. + case "UserCreated", + "UserDisabled", + "UserEnabled", + "UserLinuxUsernameChanged", + "UserProvisioningSettingsUpdated", + "UserSshSettingsUpdated", + "UserProfileUpdated", + "UserSshKeyAdded", + "UserSshKeyRemoved", + "UserEmailChanged": + // stream_type=user, stream_id=user_id + return SyncOpSyncUser, []string{e.StreamID} + + case "UserRoleAssigned", "UserRoleRevoked": + // stream_type=user_role, stream_id=user_id:role_id; user_id + // also lives in event.data["user_id"] for clarity. Prefer the + // data payload (cleaner contract); fall back to splitting the + // StreamID so a future emitter that drops the data field + // degrades to "still works" instead of "silent SyncOpNone for + // up to one reconcile interval." The role case is the only + // place this fallback is meaningful — group membership uses + // the group ID as StreamID, so user_id is data-only. + if uid := userIDFromEventData(e); uid != "" { + return SyncOpSyncUser, []string{uid} + } + if uid, _, ok := strings.Cut(e.StreamID, ":"); ok && uid != "" { + return SyncOpSyncUser, []string{uid} + } + return SyncOpNone, nil + + case "UserGroupMemberAdded", "UserGroupMemberRemoved": + // stream_type=user_group; event.data carries user_id of the + // added/removed member. StreamID format is mixed across + // emitters: SCIM (internal/scim/groups.go), the API user- + // group handler, and testutil all emit composite + // "groupID:userID" while internal/idp/linker.go uses just + // the group id. We prefer the data payload for clarity but + // fall back to splitting the StreamID for the composite- + // format emitters so a future regression that drops + // data.user_id while keeping the composite StreamID still + // produces a valid sync. + if uid := userIDFromEventData(e); uid != "" { + return SyncOpSyncUser, []string{uid} + } + // Composite "groupID:userID" → take the suffix. + if _, uid, ok := strings.Cut(e.StreamID, ":"); ok && uid != "" { + return SyncOpSyncUser, []string{uid} + } + return SyncOpNone, nil + + // UserDeleted is deliberately NOT handled here. + // CleanupDeletedUserActions needs the user projection loaded + // BEFORE the delete is applied (it reads the system_*_action_id + // columns to find the actions to clean up), so it must run in + // the handler that emits the event, not in a post-commit + // listener. DeleteUser in user_handler.go and the SCIM + // delete path each call CleanupDeletedUserActions with the + // pre-delete projection. This is the one place the derived- + // model invariant gets a documented exception. + + // Fan-out events — affect every holder / member / user. Route to + // the full sweep instead of materialising the affected set. + case "RoleUpdated", + "RoleDeleted", + "UserGroupRoleAssigned", + "UserGroupRoleRevoked", + "UserGroupDeleted", + "UserGroupQueryUpdated", + "ServerSettingUpdated": + return SyncOpSyncAll, nil + + default: + return SyncOpNone, nil + } +} + +// userIDFromEventData extracts event.data["user_id"] as a string. +// Returns "" if the field is missing or not a string — caller treats +// that as a no-op event rather than panicking. +func userIDFromEventData(e store.PersistedEvent) string { + if len(e.Data) == 0 { + return "" + } + var data map[string]any + if err := json.Unmarshal(e.Data, &data); err != nil { + return "" + } + if v, ok := data["user_id"].(string); ok { + return v + } + return "" +} + +// defaultListenerSyncTimeout is the fallback per-dispatch context +// deadline when a caller passes 0. Matches the periodic reconciler's +// default sweep timeout (5m) so a wedged DB / signer can't leak a +// goroutine on either path. Callers that want a different bound pass +// it explicitly to SystemActionListener. +const defaultListenerSyncTimeout = 5 * time.Minute + +// listenerMaxConcurrentDispatches caps the number of in-flight sync +// goroutines spawned by the listener. Prevents a SCIM bulk role +// assignment that emits thousands of UserRoleAssigned events from +// saturating the pgx pool / signer. When the cap is hit the event is +// dropped (logged) — the periodic reconciler will catch up within one +// interval, so back-pressure is preferred over unbounded fan-out. +// Round-3 review of rc11 #77. +const listenerMaxConcurrentDispatches = 16 + +// SystemActionListener is the registerable EventListener that turns +// AppendEvent post-commit hooks into system-action sync calls. Wire it +// into the store at service boot in cmd/control/main.go: +// +// st.RegisterEventListener(api.SystemActionListener(svc.SystemActions(), logger, cfg.SystemActionReconcileTimeout)) +// +// Errors from the underlying sync calls are logged and swallowed — +// listeners are post-commit, fire-and-forget; failures are caught by +// the periodic reconciler within one interval. +// +// The listener spawns a goroutine for every dispatch so AppendEvent's +// post-commit path is never blocked on system-action work. Synchronous +// invocation would have made fan-out events (RoleUpdated, RoleDeleted, +// UserGroupRoleAssigned/Revoked, UserGroupDeleted, ServerSettingUpdated) +// turn small admin RPCs into O(all-users) request-path work. +// +// Concurrency control: +// - syncTimeout bounds each goroutine (defaulting to +// defaultListenerSyncTimeout when 0). Without the bound a wedged +// DB / signer would leak goroutines indefinitely. +// - A bounded semaphore (listenerMaxConcurrentDispatches) caps +// in-flight syncs so a burst of per-user events can't exhaust +// the pgx pool. Over-cap events are dropped + logged; the +// reconciler will catch them. +// - SyncOpSyncAll uses an atomic.Bool to coalesce: if a fan-out +// sweep is already running, subsequent fan-out events return +// immediately rather than stacking N concurrent O(all-users) +// sweeps that step on each other. Same pattern the reconciler +// uses for its tick path. +// +// Each goroutine uses context.WithoutCancel(parent) under +// context.WithTimeout: the AppendEvent ctx is detached (the RPC may +// have already returned by sync time), but request-scoped values +// like request_id are preserved so error logs correlate back to the +// triggering RPC. +func SystemActionListener(mgr *SystemActionManager, logger *slog.Logger, syncTimeout time.Duration) store.EventListener { + if syncTimeout <= 0 { + syncTimeout = defaultListenerSyncTimeout + } + + sem := make(chan struct{}, listenerMaxConcurrentDispatches) + var syncAllInFlight atomic.Bool + + return func(parent context.Context, e store.PersistedEvent) { + op, userIDs := AffectedFromEvent(e) + if op == SyncOpNone { + return + } + + // Coalesce fan-out events. If a sweep is already running it + // will pick up state changes emitted before its commit; if + // not, we own the flag and must clear it on goroutine exit. + if op == SyncOpSyncAll && !syncAllInFlight.CompareAndSwap(false, true) { + logger.Debug("system-action listener: coalescing fan-out event into in-flight sweep", + "event_type", e.EventType, "event_id", e.ID) + return + } + + select { + case sem <- struct{}{}: + default: + logger.Warn("system-action listener: dispatch backpressure, dropping event (reconciler will catch up)", + "event_type", e.EventType, "event_id", e.ID, "max_concurrent", listenerMaxConcurrentDispatches) + if op == SyncOpSyncAll { + syncAllInFlight.Store(false) + } + return + } + + go func() { + defer func() { <-sem }() + if op == SyncOpSyncAll { + defer syncAllInFlight.Store(false) + } + // fireListeners' recover only protects the listener + // function itself, which returned the moment we spawned + // this goroutine. A panic from mgr.Sync* (nil deref, + // downstream library bug, etc.) without this recover + // would crash the whole control server. Round-5 review + // fix — listener API documents "post-commit + // notification" semantics, panics here must not take + // the process down. + defer func() { + if r := recover(); r != nil { + logger.Error("system-action listener: dispatch goroutine panicked", + "event_type", e.EventType, "event_id", e.ID, "panic", r) + } + }() + + ctx, cancel := context.WithTimeout(context.WithoutCancel(parent), syncTimeout) + defer cancel() + + switch op { + case SyncOpSyncUser: + for _, uid := range userIDs { + if err := mgr.SyncUserSystemActions(ctx, uid); err != nil { + logger.Error("system-action listener: sync user failed", + "user_id", uid, "event_type", e.EventType, "event_id", e.ID, "error", err) + } + } + + case SyncOpCleanupUser: + // AffectedFromEvent never returns this op currently — + // see the comment on the UserDeleted case in the + // classifier. Kept as a tagged enum for future events + // that don't have the load-before-emit ordering issue. + logger.Warn("system-action listener: SyncOpCleanupUser invoked but not implemented; handler-side cleanup is canonical", + "event_type", e.EventType, "event_id", e.ID) + + case SyncOpSyncAll: + if err := mgr.SyncAllUsersSystemActions(ctx); err != nil { + logger.Error("system-action listener: sync all users failed", + "event_type", e.EventType, "event_id", e.ID, "error", err) + } + } + }() + } +} diff --git a/internal/api/system_actions_listener_test.go b/internal/api/system_actions_listener_test.go new file mode 100644 index 00000000..c6ebdb6b --- /dev/null +++ b/internal/api/system_actions_listener_test.go @@ -0,0 +1,366 @@ +package api + +import ( + "encoding/json" + "slices" + "testing" + + "github.com/manchtools/power-manage/server/internal/store" +) + +// TestAffectedFromEvent covers the rc11 #77 derived-projection +// classifier. Each case asserts the (op, userIDs) pair AffectedFromEvent +// returns for a representative event of each handled type. Adding a +// new permission-shaping event type to the system requires extending +// AffectedFromEvent; adding the case here is the test that locks in +// the contract. +// +// Coverage rules of thumb: +// - Per-user events emit SyncOpSyncUser with one user ID. +// - Fan-out events emit SyncOpSyncAll with no user IDs. +// - UserDeleted is handled in the originating handler (load-before-emit +// ordering), so the listener returns SyncOpNone — see the comment +// on the UserDeleted case in AffectedFromEvent. +// - Unknown event types return SyncOpNone. +func TestAffectedFromEvent(t *testing.T) { + cases := []struct { + name string + event store.PersistedEvent + wantOp SyncOp + wantUsers []string + }{ + // Per-user events keyed by stream_id. + { + name: "UserCreated → sync user (stream_id)", + event: store.PersistedEvent{ + StreamType: "user", + StreamID: "user-1", + EventType: "UserCreated", + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-1"}, + }, + { + name: "UserDisabled → sync user", + event: store.PersistedEvent{ + StreamType: "user", + StreamID: "user-2", + EventType: "UserDisabled", + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-2"}, + }, + { + name: "UserLinuxUsernameChanged → sync user", + event: store.PersistedEvent{ + StreamType: "user", + StreamID: "user-3", + EventType: "UserLinuxUsernameChanged", + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-3"}, + }, + { + name: "UserSshKeyAdded → sync user", + event: store.PersistedEvent{ + StreamType: "user", + StreamID: "user-4", + EventType: "UserSshKeyAdded", + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-4"}, + }, + + // Per-user events keyed by event.data.user_id. + { + name: "UserRoleAssigned → sync user (from data.user_id)", + event: store.PersistedEvent{ + StreamType: "user_role", + StreamID: "user-5:role-x", + EventType: "UserRoleAssigned", + Data: mustMarshalJSON(t, map[string]any{"user_id": "user-5", "role_id": "role-x"}), + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-5"}, + }, + { + name: "UserRoleRevoked → sync user", + event: store.PersistedEvent{ + StreamType: "user_role", + StreamID: "user-6:role-y", + EventType: "UserRoleRevoked", + Data: mustMarshalJSON(t, map[string]any{"user_id": "user-6", "role_id": "role-y"}), + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-6"}, + }, + { + name: "UserGroupMemberAdded → sync user (from data.user_id)", + event: store.PersistedEvent{ + StreamType: "user_group", + StreamID: "group-1", + EventType: "UserGroupMemberAdded", + Data: mustMarshalJSON(t, map[string]any{"user_id": "user-7", "group_id": "group-1"}), + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-7"}, + }, + + // Fan-out events — sync everyone. + { + name: "RoleUpdated → sync all (every holder may have changed permissions)", + event: store.PersistedEvent{ + StreamType: "role", + StreamID: "role-x", + EventType: "RoleUpdated", + }, + wantOp: SyncOpSyncAll, + wantUsers: nil, + }, + { + name: "RoleDeleted → sync all", + event: store.PersistedEvent{ + StreamType: "role", + StreamID: "role-x", + EventType: "RoleDeleted", + }, + wantOp: SyncOpSyncAll, + wantUsers: nil, + }, + { + name: "UserGroupRoleAssigned → sync all (every group member affected)", + event: store.PersistedEvent{ + StreamType: "user_group", + StreamID: "group-1", + EventType: "UserGroupRoleAssigned", + }, + wantOp: SyncOpSyncAll, + wantUsers: nil, + }, + { + name: "UserGroupDeleted → sync all", + event: store.PersistedEvent{ + StreamType: "user_group", + StreamID: "group-1", + EventType: "UserGroupDeleted", + }, + wantOp: SyncOpSyncAll, + wantUsers: nil, + }, + { + name: "ServerSettingUpdated → sync all (provisioning/SSH global flip affects everyone)", + event: store.PersistedEvent{ + StreamType: "server_settings", + StreamID: "global", + EventType: "ServerSettingUpdated", + }, + wantOp: SyncOpSyncAll, + wantUsers: nil, + }, + + // Deliberate no-ops. + { + name: "UserDeleted → SyncOpNone (handler-side cleanup; see classifier comment)", + event: store.PersistedEvent{ + StreamType: "user", + StreamID: "user-99", + EventType: "UserDeleted", + }, + wantOp: SyncOpNone, + wantUsers: nil, + }, + { + name: "Unknown event type → SyncOpNone", + event: store.PersistedEvent{ + StreamType: "device", + StreamID: "device-1", + EventType: "DeviceRegistered", + }, + wantOp: SyncOpNone, + wantUsers: nil, + }, + { + name: "UserRoleAssigned with missing user_id in data → falls back to StreamID prefix", + event: store.PersistedEvent{ + StreamType: "user_role", + StreamID: "user-x:role-y", + EventType: "UserRoleAssigned", + Data: mustMarshalJSON(t, map[string]any{"role_id": "role-y"}), + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-x"}, + }, + { + name: "UserRoleRevoked with no data + no colon in StreamID → SyncOpNone", + event: store.PersistedEvent{ + StreamType: "user_role", + StreamID: "garbage-no-colon", + EventType: "UserRoleRevoked", + }, + wantOp: SyncOpNone, + wantUsers: nil, + }, + { + name: "UserGroupMemberAdded with missing user_id, single-label StreamID → SyncOpNone (no fallback target)", + event: store.PersistedEvent{ + StreamType: "user_group", + StreamID: "group-1", + EventType: "UserGroupMemberAdded", + Data: mustMarshalJSON(t, map[string]any{"group_id": "group-1"}), + }, + wantOp: SyncOpNone, + wantUsers: nil, + }, + { + name: "UserGroupMemberAdded with missing user_id, composite StreamID → falls back to suffix", + event: store.PersistedEvent{ + StreamType: "user_group", + StreamID: "group-1:user-y", + EventType: "UserGroupMemberAdded", + Data: mustMarshalJSON(t, map[string]any{"group_id": "group-1"}), + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-y"}, + }, + { + name: "UserGroupMemberRemoved with missing user_id, composite StreamID → falls back to suffix", + event: store.PersistedEvent{ + StreamType: "user_group", + StreamID: "group-2:user-z", + EventType: "UserGroupMemberRemoved", + Data: mustMarshalJSON(t, map[string]any{"group_id": "group-2"}), + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-z"}, + }, + { + name: "UserRoleAssigned with malformed JSON in Data → falls back to StreamID prefix", + event: store.PersistedEvent{ + StreamType: "user_role", + StreamID: "user-y:role-z", + EventType: "UserRoleAssigned", + // Exercises the json.Unmarshal error path in + // userIDFromEventData — round-5 review nit. + Data: []byte("{invalid"), + }, + wantOp: SyncOpSyncUser, + wantUsers: []string{"user-y"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + gotOp, gotUsers := AffectedFromEvent(tc.event) + if gotOp != tc.wantOp { + t.Errorf("op = %v, want %v", gotOp, tc.wantOp) + } + if !slices.Equal(gotUsers, tc.wantUsers) { + t.Errorf("users = %v, want %v", gotUsers, tc.wantUsers) + } + }) + } +} + +// TestAffectedFromEvent_PerLiteral locks every event-type literal +// handled by AffectedFromEvent's switch to a specific SyncOp outcome. +// The structured cases above cover all four SyncOp branches, but +// several literals share a case and would silently flip to SyncOpNone +// if a future refactor accidentally drops one from the case list. +// Iterating the literals individually catches that regression. +func TestAffectedFromEvent_PerLiteral(t *testing.T) { + // Per-user events keyed by stream_id (stream_type=user). Bare + // PersistedEvent with the literal as EventType + a fixed user id + // is enough — the classifier reads no other field for these. + syncUserStreamLiterals := []string{ + "UserCreated", + "UserDisabled", + "UserEnabled", + "UserLinuxUsernameChanged", + "UserProvisioningSettingsUpdated", + "UserSshSettingsUpdated", + "UserProfileUpdated", + "UserSshKeyAdded", + "UserSshKeyRemoved", + "UserEmailChanged", + } + for _, et := range syncUserStreamLiterals { + t.Run("user_stream/"+et, func(t *testing.T) { + op, users := AffectedFromEvent(store.PersistedEvent{ + StreamType: "user", + StreamID: "user-x", + EventType: et, + }) + if op != SyncOpSyncUser { + t.Errorf("op = %v, want SyncOpSyncUser", op) + } + if !slices.Equal(users, []string{"user-x"}) { + t.Errorf("users = %v, want [user-x]", users) + } + }) + } + + // Per-user events keyed by event.data.user_id (with StreamID + // fallback for the role pair). + dataPayload := mustMarshalJSON(t, map[string]any{"user_id": "user-y"}) + syncUserDataLiterals := []struct { + EventType string + StreamType string + StreamID string + }{ + {"UserRoleAssigned", "user_role", "user-y:role-z"}, + {"UserRoleRevoked", "user_role", "user-y:role-z"}, + {"UserGroupMemberAdded", "user_group", "group-1"}, + {"UserGroupMemberRemoved", "user_group", "group-1"}, + } + for _, c := range syncUserDataLiterals { + t.Run("user_data/"+c.EventType, func(t *testing.T) { + op, users := AffectedFromEvent(store.PersistedEvent{ + StreamType: c.StreamType, + StreamID: c.StreamID, + EventType: c.EventType, + Data: dataPayload, + }) + if op != SyncOpSyncUser { + t.Errorf("op = %v, want SyncOpSyncUser", op) + } + if !slices.Equal(users, []string{"user-y"}) { + t.Errorf("users = %v, want [user-y]", users) + } + }) + } + + // Fan-out events. + syncAllLiterals := []string{ + "RoleUpdated", + "RoleDeleted", + "UserGroupRoleAssigned", + "UserGroupRoleRevoked", + "UserGroupDeleted", + "UserGroupQueryUpdated", + "ServerSettingUpdated", + } + for _, et := range syncAllLiterals { + t.Run("sync_all/"+et, func(t *testing.T) { + op, users := AffectedFromEvent(store.PersistedEvent{ + StreamType: "_unused_", + StreamID: "_unused_", + EventType: et, + }) + if op != SyncOpSyncAll { + t.Errorf("op = %v, want SyncOpSyncAll", op) + } + if users != nil { + t.Errorf("users = %v, want nil", users) + } + }) + } +} + +func mustMarshalJSON(t *testing.T, v any) []byte { + t.Helper() + b, err := json.Marshal(v) + if err != nil { + t.Fatalf("marshal: %v", err) + } + return b +} + diff --git a/internal/api/terminal_handler.go b/internal/api/terminal_handler.go index 79e10963..64093c56 100644 --- a/internal/api/terminal_handler.go +++ b/internal/api/terminal_handler.go @@ -264,15 +264,93 @@ func (h *TerminalHandler) resolveGatewayURL(ctx context.Context, deviceID string terminalURL, err := h.registry.LookupGatewayTerminalURL(ctx, gatewayID) if err != nil { if errors.Is(err, registry.ErrNoGateway) { - // The gateway died between the device lookup and the - // URL lookup. Surface as Unavailable so the client - // retries — by then the agent has reconnected - // elsewhere and the next StartTerminal call resolves - // to a live gateway. - h.logger.Warn("gateway hosting device is no longer registered", - "device_id", deviceID, "gateway_id", gatewayID) + // Two distinct causes surface as ErrNoGateway here: + // + // 1. Transient — the gateway died between the device + // lookup and the URL lookup. Retry recovers once + // the agent reconnects elsewhere. + // 2. Persistent — the gateway is up and the device is + // connected to it, but the gateway never published + // its terminal URL because + // GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE is unset. + // The rc10 staging failure mode. + // + // Probe the internal-URL key for the same gatewayID to + // distinguish: if internal is present and terminal is + // not, the gateway is alive and the missing terminal URL + // is persistent operator-facing misconfig + // (ErrGatewayNotRegistered with the "set the env var" + // message). If neither is present the gateway is gone — + // surface that as ErrDeviceNotConnected so the web + // suggests retry rather than reporting a config error + // for a transient race. + _, internalErr := h.registry.LookupGatewayInternalURL(ctx, gatewayID) + if internalErr == nil { + // The terminal-URL and internal-URL keys are + // refreshed by independent goroutines on the + // gateway side (registry.RegisterGateway and + // cmd/gateway/main.go's RegisterGatewayInternal + // loop). Same TTL, same refresh interval, but the + // tickers drift — so the terminal-URL key can + // expire ~tens of ms before the internal-URL key + // just from natural skew. Without dampening that + // race surfaces as ErrGatewayNotRegistered (the + // persistent-misconfig branch) and pages operators + // for what's actually transient. Retry the + // terminal-URL lookup briefly before classifying + // as persistent — round-6 review fix on rc11 #79. + const ( + driftRetries = 3 + driftBackoff = 30 * time.Millisecond + ) + for i := 0; i < driftRetries; i++ { + select { + case <-ctx.Done(): + return "", apiErrorCtx(ctx, ErrInternal, connect.CodeUnavailable, + "context cancelled during gateway terminal URL retry") + case <-time.After(driftBackoff): + } + recoveredURL, retryErr := h.registry.LookupGatewayTerminalURL(ctx, gatewayID) + if retryErr == nil { + h.logger.Debug("gateway terminal URL recovered after retry (refresh-goroutine drift bridged)", + "device_id", deviceID, "gateway_id", gatewayID, "attempt", i+1) + return recoveredURL, nil + } + if !errors.Is(retryErr, registry.ErrNoGateway) { + // A different registry error during retry + // (e.g. backend unreachable) — fall through + // to the persistent classification below + // rather than retrying further on a + // non-recoverable failure mode. + h.logger.Warn("gateway terminal URL retry failed with non-ErrNoGateway error", + "device_id", deviceID, "gateway_id", gatewayID, "error", retryErr) + break + } + } + + // Case 2 — gateway alive, terminal URL still missing + // after the drift-tolerant retry window. Persistent + // misconfiguration. + h.logger.Warn("gateway hosting device has no terminal URL registered (alive, terminal-URL not published)", + "device_id", deviceID, "gateway_id", gatewayID) + return "", apiErrorCtx(ctx, ErrGatewayNotRegistered, connect.CodeUnavailable, + "gateway hosting this device has not published its terminal URL — ensure GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE is set on the gateway") + } + if errors.Is(internalErr, registry.ErrNoGateway) { + // Case 1 — gateway gone. Transient; UI should + // suggest retry rather than imply config error. + h.logger.Info("gateway hosting device disappeared between lookups (transient)", + "device_id", deviceID, "gateway_id", gatewayID) + return "", apiErrorCtx(ctx, ErrDeviceNotConnected, connect.CodeUnavailable, + "device's gateway is no longer connected; retry shortly") + } + // Internal-URL probe failed for a different reason + // (registry unreachable, etc.). Fall through to the + // generic registry-unavailable path below. + h.logger.Error("registry probe failed when distinguishing persistent vs transient gateway loss", + "device_id", deviceID, "gateway_id", gatewayID, "internal_probe_error", internalErr) return "", apiErrorCtx(ctx, ErrInternal, connect.CodeUnavailable, - "gateway hosting this device is no longer registered; retry shortly") + "registry lookup failed; retry shortly") } h.logger.Error("gateway URL lookup failed", "gateway_id", gatewayID, "error", err) diff --git a/internal/api/terminal_handler_test.go b/internal/api/terminal_handler_test.go index 0fbcdb1f..8f69851d 100644 --- a/internal/api/terminal_handler_test.go +++ b/internal/api/terminal_handler_test.go @@ -20,6 +20,21 @@ import ( "github.com/manchtools/power-manage/server/internal/testutil" ) +// errorCode pulls the structured error code (pm.ErrorDetail.Code) +// off a connect.Error. The SDK + web client both switch on this +// field, so it's the canonical contract test for the rc11 #79 +// terminal error split — message substring assertions are a secondary +// hint check that's free to evolve with copy edits. +func errorCode(t *testing.T, e *connect.Error) string { + t.Helper() + require.NotEmpty(t, e.Details(), "connect error has no structured details — apiErrorCtx wiring broken?") + val, err := e.Details()[0].Value() + require.NoError(t, err, "decode ErrorDetail proto") + detail, ok := val.(*pm.ErrorDetail) + require.True(t, ok, "first detail is not pm.ErrorDetail (got %T)", val) + return detail.Code +} + // setLinuxUsername appends the UserLinuxUsernameChanged event so the // projection picks up the linux_username for tests that need // StartTerminal to resolve a TTY user. @@ -346,3 +361,98 @@ func TestStartTerminal_RegistryDeviceNotConnected(t *testing.T) { require.True(t, errors.As(err, &connectErr)) assert.Equal(t, connect.CodeFailedPrecondition, connectErr.Code()) } + +// TestStartTerminal_GatewayNotRegistered_PersistentMisconfig covers +// the rc11 #79 split: when the device→gateway mapping exists AND the +// gateway has registered its internal URL but NOT a terminal URL, +// resolveGatewayURL distinguishes this persistent operator-facing +// misconfig (operator forgot GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE) +// from a transient gateway-gone race. Persistent → ErrGatewayNotRegistered +// with the actionable "set the env" message. +func TestStartTerminal_GatewayNotRegistered_PersistentMisconfig(t *testing.T) { + st := testutil.SetupPostgres(t) + tokenStore := terminal.NewTokenStore(terminal.NewFakeBackend(nil)) + reg := registry.New(registry.NewFakeBackend(nil), slog.Default()) + h := api.NewTerminalHandler(st, tokenStore, reg, "", slog.Default()) + + userID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") + setLinuxUsername(t, st, userID, "alice") + deviceID := testutil.CreateTestDevice(t, st, "host-misconfig") + testutil.AssignDeviceToUser(t, st, userID, deviceID, userID) + + // Simulate the rc10 staging failure mode: agent is connected + // (device→gateway mapping exists), gateway has published its + // internal URL (alive, control can fan out admin RPCs), but the + // terminal URL is missing because the operator never set + // GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE on the gateway. + ctx := context.Background() + require.NoError(t, reg.AttachDevice(ctx, deviceID, "gw-misconfig", registry.DefaultDeviceTTL)) + require.NoError(t, reg.RegisterGatewayInternal(ctx, "gw-misconfig", + "https://gw-misconfig.internal:8080", registry.DefaultGatewayTTL)) + // Note: NO RegisterGateway call — terminal URL never published. + + _, err := h.StartTerminal(authedCtx(userID), connect.NewRequest(&pm.StartTerminalRequest{ + DeviceId: deviceID, + })) + require.Error(t, err) + var connectErr *connect.Error + require.True(t, errors.As(err, &connectErr)) + // Unavailable gRPC code (transient-style retry semantics for the + // connection layer), with the persistent error code so the web + // client can show the operator-actionable message. Asserting the + // structured Code is the primary contract — the SDK + web client + // switch on it via ErrorDetail.findDetails. Message substring is + // kept as a secondary check on the operator hint. + assert.Equal(t, connect.CodeUnavailable, connectErr.Code()) + assert.Equal(t, api.ErrGatewayNotRegistered, errorCode(t, connectErr), + "persistent-misconfig path must surface ErrGatewayNotRegistered, not the transient code") + assert.Contains(t, connectErr.Error(), "GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE", + "persistent-misconfig path must name the missing env var so the operator can fix it") +} + +// TestStartTerminal_GatewayNotRegistered_TransientGatewayLoss covers +// the other half of the rc11 #79 split: when the device→gateway +// mapping briefly exists but BOTH the gateway's internal URL and +// terminal URL are missing (gateway disappeared between the device +// lookup and the URL lookup), resolveGatewayURL surfaces this as +// ErrDeviceNotConnected with retry semantics — not as a config +// error. The previous cut conflated this with the persistent +// misconfig case and produced the wrong UI message. +func TestStartTerminal_GatewayNotRegistered_TransientGatewayLoss(t *testing.T) { + st := testutil.SetupPostgres(t) + tokenStore := terminal.NewTokenStore(terminal.NewFakeBackend(nil)) + reg := registry.New(registry.NewFakeBackend(nil), slog.Default()) + h := api.NewTerminalHandler(st, tokenStore, reg, "", slog.Default()) + + userID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") + setLinuxUsername(t, st, userID, "alice") + deviceID := testutil.CreateTestDevice(t, st, "host-transient") + testutil.AssignDeviceToUser(t, st, userID, deviceID, userID) + + // Simulate the transient race: device→gateway mapping exists, + // but the gateway has no published URLs at all (it died, or + // hasn't refreshed yet). The internal-URL probe also returns + // ErrNoGateway → the resolver classifies this as transient. + ctx := context.Background() + require.NoError(t, reg.AttachDevice(ctx, deviceID, "gw-gone", registry.DefaultDeviceTTL)) + // Note: NEITHER RegisterGateway nor RegisterGatewayInternal — + // gateway is fully gone from the registry's URL keys. + + _, err := h.StartTerminal(authedCtx(userID), connect.NewRequest(&pm.StartTerminalRequest{ + DeviceId: deviceID, + })) + require.Error(t, err) + var connectErr *connect.Error + require.True(t, errors.As(err, &connectErr)) + assert.Equal(t, connect.CodeUnavailable, connectErr.Code()) + assert.Equal(t, api.ErrDeviceNotConnected, errorCode(t, connectErr), + "transient gateway-loss path must surface ErrDeviceNotConnected, not the persistent code") + // The transient path must NOT surface the "set the env var" hint. + // The negative assertion is the high-signal contract test — + // CodeUnavailable + ErrDeviceNotConnected together already encode + // "transient, retryable" for callers, so we don't pin the exact + // wording of the operator-facing message (it's free to evolve via + // copy edits without breaking the SDK contract). + assert.NotContains(t, connectErr.Error(), "GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE", + "transient gateway-loss path must not surface a config-error message") +} diff --git a/internal/api/user_handler.go b/internal/api/user_handler.go index f2be0238..13cc9de7 100644 --- a/internal/api/user_handler.go +++ b/internal/api/user_handler.go @@ -172,12 +172,10 @@ func (h *UserHandler) CreateUser(ctx context.Context, req *connect.Request[pm.Cr h.enqueueUserReindex(ctx, user) - // Sync system actions (fire-and-forget) - if h.systemActions != nil { - if err := h.systemActions.SyncUserSystemActions(ctx, id); err != nil { - h.logger.Error("failed to sync system actions after user creation", "user_id", id, "error", err) - } - } + // System-action sync runs from the post-commit listener + // registered on the store (see api.SystemActionListener) — handler- + // side calls were removed in rc11 #77 to keep the derived-model + // invariant: handlers mutate source state only. protoUser := userToProto(user) h.populateUserRoles(ctx, protoUser) @@ -410,12 +408,7 @@ func (h *UserHandler) SetUserDisabled(ctx context.Context, req *connect.Request[ h.enqueueUserReindex(ctx, user) - // Sync system actions (disabled flag changes USER action params) - if h.systemActions != nil { - if err := h.systemActions.SyncUserSystemActions(ctx, req.Msg.Id); err != nil { - h.logger.Error("failed to sync system actions after disable/enable", "user_id", req.Msg.Id, "error", err) - } - } + // System-action sync runs from the post-commit listener (rc11 #77). return connect.NewResponse(&pm.UpdateUserResponse{ User: userToProto(user), @@ -509,12 +502,7 @@ func (h *UserHandler) UpdateUserProfile(ctx context.Context, req *connect.Reques h.enqueueUserReindex(ctx, user) - // Sync system actions (display_name change affects USER action comment) - if h.systemActions != nil { - if err := h.systemActions.SyncUserSystemActions(ctx, req.Msg.Id); err != nil { - h.logger.Error("failed to sync system actions after profile update", "user_id", req.Msg.Id, "error", err) - } - } + // System-action sync runs from the post-commit listener (rc11 #77). return connect.NewResponse(&pm.UpdateUserResponse{ User: userToProto(user), @@ -602,10 +590,7 @@ func (h *UserHandler) SetUserProvisioningEnabled(ctx context.Context, req *conne return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to update user provisioning settings") } - // Sync system actions - if err := h.systemActions.SyncUserSystemActions(ctx, req.Msg.UserId); err != nil { - h.logger.Error("failed to sync system actions after provisioning toggle", "user_id", req.Msg.UserId, "error", err) - } + // System-action sync runs from the post-commit listener (rc11 #77). user, err := h.store.Queries().GetUserByID(ctx, req.Msg.UserId) if err != nil { @@ -663,11 +648,7 @@ func (h *UserHandler) UpdateUserLinuxUsername(ctx context.Context, req *connect. h.enqueueUserReindex(ctx, user) - if h.systemActions != nil { - if err := h.systemActions.SyncUserSystemActions(ctx, req.Msg.UserId); err != nil { - h.logger.Error("failed to sync system actions after username change", "user_id", req.Msg.UserId, "error", err) - } - } + // System-action sync runs from the post-commit listener (rc11 #77). return connect.NewResponse(&pm.UpdateUserResponse{ User: userToProto(user), @@ -709,11 +690,7 @@ func (h *UserHandler) AddUserSshKey(ctx context.Context, req *connect.Request[pm return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to add SSH key") } - if h.systemActions != nil { - if err := h.systemActions.SyncUserSystemActions(ctx, req.Msg.UserId); err != nil { - h.logger.Error("failed to sync system actions after SSH key added", "user_id", req.Msg.UserId, "error", err) - } - } + // System-action sync runs from the post-commit listener (rc11 #77). return connect.NewResponse(&pm.AddUserSshKeyResponse{ Key: &pm.SshPublicKey{ @@ -754,11 +731,7 @@ func (h *UserHandler) RemoveUserSshKey(ctx context.Context, req *connect.Request return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to remove SSH key") } - if h.systemActions != nil { - if err := h.systemActions.SyncUserSystemActions(ctx, req.Msg.UserId); err != nil { - h.logger.Error("failed to sync system actions after SSH key removed", "user_id", req.Msg.UserId, "error", err) - } - } + // System-action sync runs from the post-commit listener (rc11 #77). return connect.NewResponse(&pm.RemoveUserSshKeyResponse{}), nil } @@ -799,11 +772,7 @@ func (h *UserHandler) UpdateUserSshSettings(ctx context.Context, req *connect.Re return nil, handleGetError(ctx, err, ErrUserNotFound, "user not found") } - if h.systemActions != nil { - if err := h.systemActions.SyncUserSystemActions(ctx, req.Msg.UserId); err != nil { - h.logger.Error("failed to sync system actions after SSH settings update", "user_id", req.Msg.UserId, "error", err) - } - } + // System-action sync runs from the post-commit listener (rc11 #77). return connect.NewResponse(&pm.UpdateUserResponse{ User: userToProto(user), diff --git a/internal/config/config.go b/internal/config/config.go index 49c23b1f..390077ba 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "strconv" + "strings" "time" ) @@ -146,6 +147,14 @@ type Config struct { // Logging LogLevel string + + // TTYDomainExplicitlySet captures whether GATEWAY_TTY_DOMAIN was + // set in the environment, separate from the always-falls-back- + // to-GATEWAY_DOMAIN behavior of TraefikTTYHost. Used by Validate + // to surface partial terminal-config combinations as warnings + // (#78). False does not mean the TTY domain is unused — the + // Traefik routing fallback still produces a usable host. + TTYDomainExplicitlySet bool } // FromEnv loads configuration from environment variables. @@ -219,32 +228,38 @@ func FromEnv() *Config { TraefikTTYCertResolver: getEnv("GATEWAY_TRAEFIK_TTY_CERT_RESOLVER", "letsencrypt"), HeartbeatInterval: getEnvHeartbeatInterval("GATEWAY_HEARTBEAT_INTERVAL"), LogLevel: getEnv("GATEWAY_LOG_LEVEL", "info"), + // Explicit-set tracking for partial-terminal-config warning + // in Validate; see field doc above. + TTYDomainExplicitlySet: os.Getenv("GATEWAY_TTY_DOMAIN") != "", } } -// Validate returns a non-nil error when the loaded config has a -// combination that the gateway cannot serve coherently. Called once -// at startup from cmd/gateway/main.go; keeping it on the Config -// struct (rather than inline in main) so tests can exercise the -// shape checks without booting a full process. -func (c *Config) Validate() error { - // TTY / MTLS host collision: when Traefik self-registration is - // on AND the terminal feature is in play AND the mTLS + TTY +// Validate returns warnings the operator should see at startup and a +// non-nil error when the loaded config has a combination that the +// gateway cannot serve coherently. Called once at startup from +// cmd/gateway/main.go; keeping it on the Config struct (rather than +// inline in main) so tests can exercise the shape checks without +// booting a full process. +// +// Warnings are non-fatal — the gateway still starts. They surface +// partial misconfigurations that would otherwise produce silent +// runtime failures (typical example: an operator who set +// GATEWAY_WEB_LISTEN_ADDR but forgot GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE, +// so the terminal listener exists locally but the registry never +// learns about it and StartTerminal RPC fails opaquely later). +// +// Errors are fatal — the gateway must not start. Reserved for +// combinations that produce dispatch-shadowing or silent breakage +// of the primary mTLS path. +func (c *Config) Validate() (warnings []string, err error) { + // TTY / MTLS host collision (fatal): when Traefik self-registration + // is on AND the terminal feature is in play AND the mTLS + TTY // routers bind the same entrypoint, the shared hostname means // Traefik's TCP-passthrough router for mTLS wins the SNI match // and the TTY HTTP router never gets the request. The TLS // handshake lands on an agent-mTLS backend that isn't speaking // HTTP, so the WebSocket upgrade fails silently. // - // Split the preconditions explicitly so the check handles both - // the auto-derived backend case (operator only set - // GATEWAY_WEB_LISTEN_ADDR) and the explicit backend case - // (operator set GATEWAY_TRAEFIK_TTY_BACKEND directly). Also - // narrow to "same entrypoint" — different entrypoints mean the - // routers don't actually collide even on a shared host, so - // flagging that shape would be a false positive for - // bring-your-own-Traefik topologies. - // // Refuse startup with a clear message instead of letting the // operator discover this when a terminal session silently fails. terminalEnabled := c.WebListenAddr != "" || c.TraefikTTYBackend != "" @@ -255,12 +270,49 @@ func (c *Config) Validate() error { c.TraefikTTYHost != "" && c.TraefikMTLSHost != "" && c.TraefikTTYHost == c.TraefikMTLSHost { - return fmt.Errorf( + return warnings, fmt.Errorf( "GATEWAY_TTY_DOMAIN / TraefikTTYHost cannot equal GATEWAY_DOMAIN / TraefikMTLSHost when the terminal router and mTLS router share an entrypoint (both %q on entrypoint %q): Traefik TCP passthrough for mTLS matches the same SNI as the TTY HTTP router and breaks WebSocket sessions — set a distinct GATEWAY_TTY_DOMAIN or split the entrypoints", c.TraefikTTYHost, c.TraefikTTYEntryPoint, ) } - return nil + + // Partial terminal configuration (warning, #78): the terminal + // feature is gated on three independent env vars working in + // concert — GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE (controls + // whether the gateway publishes its terminal URL to the + // registry), GATEWAY_WEB_LISTEN_ADDR (the local TTY HTTP + // listener), and GATEWAY_TTY_DOMAIN (the public hostname Traefik + // routes to the listener). Setting any one of them signals + // operator intent to enable terminal sessions. Setting some but + // not all produces a working-looking gateway that silently fails + // at StartTerminal time — exactly the rc10 staging failure mode + // this issue addresses. + // + // All-unset is the deliberate "terminal feature off" path and + // stays silent. All-set is the working path. Anything in between + // gets a warning naming the missing piece(s) so the operator + // doesn't need to spelunk Valkey / control logs to diagnose. + intent := c.WebListenAddr != "" || c.PublicTerminalURLTemplate != "" || c.TTYDomainExplicitlySet + if intent { + var missing []string + if c.PublicTerminalURLTemplate == "" { + missing = append(missing, "GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE") + } + if c.WebListenAddr == "" { + missing = append(missing, "GATEWAY_WEB_LISTEN_ADDR") + } + if !c.TTYDomainExplicitlySet { + missing = append(missing, "GATEWAY_TTY_DOMAIN") + } + if len(missing) > 0 { + warnings = append(warnings, fmt.Sprintf( + "terminal sessions appear partially configured: missing %s — set all three to enable terminals end-to-end, or unset the others to disable the feature", + strings.Join(missing, ", "), + )) + } + } + + return warnings, nil } // firstNonEmpty returns the first argument that isn't the empty diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 686858fb..711b1e75 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1,6 +1,7 @@ package config import ( + "strings" "testing" ) @@ -209,7 +210,7 @@ func TestValidate_TTYMTLSHostCollision(t *testing.T) { } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - err := tc.cfg.Validate() + _, err := tc.cfg.Validate() if tc.wantErr && err == nil { t.Fatalf("Validate() = nil, want error for %+v", tc.cfg) } @@ -220,6 +221,118 @@ func TestValidate_TTYMTLSHostCollision(t *testing.T) { } } +// TestValidate_PartialTerminalConfig covers the rc11 #78 hardening: +// when an operator sets some but not all of the three terminal env +// vars (GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE, GATEWAY_WEB_LISTEN_ADDR, +// GATEWAY_TTY_DOMAIN), Validate emits a warning naming the missing +// piece(s). All-unset is the deliberate "feature off" path and stays +// silent. All-set is the working path. Anything in between gets a +// warning so the operator can diagnose without reading control logs +// when StartTerminal later fails opaquely. +func TestValidate_PartialTerminalConfig(t *testing.T) { + cases := []struct { + name string + cfg Config + wantWarn bool + wantMissing []string // substring(s) the warning must contain when wantWarn is true + }{ + { + name: "all three unset → silent (feature off)", + cfg: Config{}, + wantWarn: false, + }, + { + name: "all three set → silent (working config)", + cfg: Config{ + PublicTerminalURLTemplate: "wss://tty.example.com/gw/{id}/terminal", + WebListenAddr: ":8443", + TTYDomainExplicitlySet: true, + }, + wantWarn: false, + }, + { + name: "WebListenAddr only → warn missing TEMPLATE + DOMAIN", + cfg: Config{ + WebListenAddr: ":8443", + }, + wantWarn: true, + wantMissing: []string{"GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE", "GATEWAY_TTY_DOMAIN"}, + }, + { + name: "TEMPLATE only → warn missing WebListenAddr + DOMAIN", + cfg: Config{ + PublicTerminalURLTemplate: "wss://tty.example.com/gw/{id}/terminal", + }, + wantWarn: true, + wantMissing: []string{"GATEWAY_WEB_LISTEN_ADDR", "GATEWAY_TTY_DOMAIN"}, + }, + { + name: "DOMAIN only → warn missing TEMPLATE + WebListenAddr", + cfg: Config{ + TTYDomainExplicitlySet: true, + }, + wantWarn: true, + wantMissing: []string{"GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE", "GATEWAY_WEB_LISTEN_ADDR"}, + }, + { + name: "TEMPLATE + WebListenAddr → warn missing DOMAIN", + cfg: Config{ + PublicTerminalURLTemplate: "wss://tty.example.com/gw/{id}/terminal", + WebListenAddr: ":8443", + }, + wantWarn: true, + wantMissing: []string{"GATEWAY_TTY_DOMAIN"}, + }, + { + name: "TEMPLATE + DOMAIN → warn missing WebListenAddr", + cfg: Config{ + PublicTerminalURLTemplate: "wss://tty.example.com/gw/{id}/terminal", + TTYDomainExplicitlySet: true, + }, + wantWarn: true, + wantMissing: []string{"GATEWAY_WEB_LISTEN_ADDR"}, + }, + { + name: "WebListenAddr + DOMAIN → warn missing TEMPLATE", + cfg: Config{ + WebListenAddr: ":8443", + TTYDomainExplicitlySet: true, + }, + wantWarn: true, + wantMissing: []string{"GATEWAY_PUBLIC_TERMINAL_URL_TEMPLATE"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + warnings, err := tc.cfg.Validate() + if err != nil { + t.Fatalf("Validate() returned err = %v, want nil (partial config is warning-only)", err) + } + if tc.wantWarn { + if len(warnings) == 0 { + t.Fatalf("Validate() warnings = empty, want a partial-terminal-config warning for %+v", tc.cfg) + } + for _, want := range tc.wantMissing { + found := false + for _, w := range warnings { + if strings.Contains(w, want) { + found = true + break + } + } + if !found { + t.Errorf("Validate() warnings = %v, want one to contain %q", warnings, want) + } + } + } else { + if len(warnings) > 0 { + t.Errorf("Validate() warnings = %v, want empty for %+v", warnings, tc.cfg) + } + } + }) + } +} + func TestGetEnvInt_ValidValues(t *testing.T) { tests := []struct { name string diff --git a/internal/scim/handler.go b/internal/scim/handler.go index 8fe21437..5aaa4148 100644 --- a/internal/scim/handler.go +++ b/internal/scim/handler.go @@ -8,6 +8,7 @@ import ( "github.com/manchtools/power-manage/server/internal/auth" "github.com/manchtools/power-manage/server/internal/store" + db "github.com/manchtools/power-manage/server/internal/store/generated" ) // appendEvent appends an event and logs any error. @@ -24,20 +25,40 @@ type contextKey string // providerContextKey is the context key for the authenticated SCIM provider. const providerContextKey contextKey = "scim_provider" +// SystemActionsCleaner is the narrow surface SCIM needs from the +// api.SystemActionManager so it can clean up system actions when a +// user is deleted via the SCIM provisioning path. Defined here as +// an interface (rather than importing api) to keep the SCIM package +// import graph free of api → scim cycles. Provided at construction +// time; nil disables cleanup (handler-test friendly). +// +// rc11 #77: SCIM was previously bypassing the system-actions +// cleanup entirely on user deletion, leaving orphan pm-tty-* and +// USER provision actions on devices. +type SystemActionsCleaner interface { + CleanupDeletedUserActions(ctx context.Context, user db.UsersProjection) error +} + // Handler handles SCIM v2 API requests. type Handler struct { - store *store.Store - logger *slog.Logger - rateLimiter *auth.RateLimiter + store *store.Store + logger *slog.Logger + rateLimiter *auth.RateLimiter + systemActions SystemActionsCleaner // optional; nil = no cleanup on delete } // NewHandler creates an http.Handler that serves all SCIM v2 routes. // Routes are mounted at /scim/v2/{slug}/... -func NewHandler(st *store.Store, logger *slog.Logger) http.Handler { +// +// systemActions may be nil — used by tests that don't exercise the +// delete cleanup path. Production wiring in cmd/control/main.go +// passes the live SystemActionManager. +func NewHandler(st *store.Store, logger *slog.Logger, systemActions SystemActionsCleaner) http.Handler { h := &Handler{ - store: st, - logger: logger, - rateLimiter: auth.NewRateLimiter(100, 1*time.Minute), + store: st, + logger: logger, + rateLimiter: auth.NewRateLimiter(100, 1*time.Minute), + systemActions: systemActions, } mux := http.NewServeMux() diff --git a/internal/scim/handler_test.go b/internal/scim/handler_test.go index 634d67ab..2d9520cb 100644 --- a/internal/scim/handler_test.go +++ b/internal/scim/handler_test.go @@ -2,10 +2,12 @@ package scim_test import ( "bytes" + "context" "encoding/json" "log/slog" "net/http" "net/http/httptest" + "sync" "testing" "github.com/stretchr/testify/assert" @@ -14,9 +16,48 @@ import ( "github.com/manchtools/power-manage/server/internal/crypto" "github.com/manchtools/power-manage/server/internal/scim" "github.com/manchtools/power-manage/server/internal/store" + db "github.com/manchtools/power-manage/server/internal/store/generated" "github.com/manchtools/power-manage/server/internal/testutil" ) +// fakeSystemActionsCleaner records each CleanupDeletedUserActions +// call so tests can assert the rc11 #77 SCIM-side cleanup is wired +// up correctly. The interface lives on scim.SystemActionsCleaner; +// the SCIM handler only calls the cleaner when the *last* identity +// link is removed and the user projection load succeeded — so a +// regression that drops either guard is observable through the +// recorded call list. +type fakeSystemActionsCleaner struct { + mu sync.Mutex + calls []db.UsersProjection +} + +func (f *fakeSystemActionsCleaner) CleanupDeletedUserActions(_ context.Context, u db.UsersProjection) error { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, u) + return nil +} + +func (f *fakeSystemActionsCleaner) callCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.calls) +} + +func (f *fakeSystemActionsCleaner) lastCall() db.UsersProjection { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.calls) == 0 { + // Caller forgot to assert callCount > 0 first. Return the + // zero value so the resulting test failure has a readable + // "want X, got " diff instead of a + // panic stack trace. + return db.UsersProjection{} + } + return f.calls[len(f.calls)-1] +} + // scimTestEnv holds shared state for SCIM integration tests. type scimTestEnv struct { handler http.Handler @@ -26,14 +67,28 @@ type scimTestEnv struct { providerID string slug string token string + cleaner *fakeSystemActionsCleaner // nil when setupSCIM was used (cleanup path not wired) } func setupSCIM(t *testing.T) *scimTestEnv { + t.Helper() + return setupSCIMWithCleaner(t, nil) +} + +// setupSCIMWithCleaner injects a SystemActionsCleaner into the SCIM +// handler so tests can observe rc11 #77's SCIM-delete cleanup path. +// Pass nil for the same behaviour as setupSCIM (no cleaner wired). +func setupSCIMWithCleaner(t *testing.T, cleaner *fakeSystemActionsCleaner) *scimTestEnv { t.Helper() st := testutil.SetupPostgres(t) enc := testutil.NewEncryptor(t) logger := slog.New(slog.NewTextHandler(&bytes.Buffer{}, nil)) - handler := scim.NewHandler(st, logger) + + var sysCleaner scim.SystemActionsCleaner + if cleaner != nil { + sysCleaner = cleaner + } + handler := scim.NewHandler(st, logger, sysCleaner) adminID := testutil.CreateTestUser(t, st, testutil.NewID()+"@test.com", "pass", "admin") slug := "scim-test-" + testutil.NewID()[:8] @@ -48,6 +103,7 @@ func setupSCIM(t *testing.T) *scimTestEnv { providerID: providerID, slug: slug, token: token, + cleaner: cleaner, } } @@ -320,6 +376,43 @@ func TestDeleteUser_Success(t *testing.T) { assert.Equal(t, http.StatusNoContent, w.Code) } +// TestDeleteUser_LastLink_TriggersSystemActionsCleanup is the +// regression test for rc11 #77's SCIM-side cleanup wiring. Before +// that fix, SCIM deleted users without ever calling +// CleanupDeletedUserActions, leaking pm-tty-* and USER provision +// actions on every device the user was assigned to. Asserting via a +// fake cleaner pins the contract that: +// +// 1. CleanupDeletedUserActions IS called when the last identity +// link is removed (the user is being fully deleted). +// 2. It receives the projection loaded BEFORE the UserDeleted event +// was emitted (so the system_*_action_id fields are still +// populated for the cleaner to read). +func TestDeleteUser_LastLink_TriggersSystemActionsCleanup(t *testing.T) { + cleaner := &fakeSystemActionsCleaner{} + env := setupSCIMWithCleaner(t, cleaner) + + user := map[string]any{ + "schemas": []string{"urn:ietf:params:scim:schemas:core:2.0:User"}, + "userName": "cleanup-target@example.com", + "externalId": "ext-cleanup-target", + } + createResp := env.request("POST", "/Users", user) + require.Equal(t, http.StatusCreated, createResp.Code) + + var created map[string]any + require.NoError(t, json.Unmarshal(createResp.Body.Bytes(), &created)) + userID := created["id"].(string) + + w := env.request("DELETE", "/Users/"+userID) + require.Equal(t, http.StatusNoContent, w.Code) + + require.Equal(t, 1, cleaner.callCount(), + "CleanupDeletedUserActions should fire exactly once on last-link DELETE; rc11 #77 regression if 0") + assert.Equal(t, userID, cleaner.lastCall().ID, + "cleaner must receive the deleted user's projection so it can read system_*_action_id columns") +} + // --- Group Tests --- func TestCreateGroup_Success(t *testing.T) { diff --git a/internal/scim/users.go b/internal/scim/users.go index 452c69ed..ac75ac47 100644 --- a/internal/scim/users.go +++ b/internal/scim/users.go @@ -846,6 +846,14 @@ func (h *Handler) deleteUser(w http.ResponseWriter, r *http.Request) { // linkCount reflects state after the IdentityUnlinked event projection. // If 0 remaining links, safe to delete the user. if linkCount == 0 { + // Load the user projection BEFORE emitting UserDeleted so + // CleanupDeletedUserActions can read the system_*_action_id + // columns that the deletion projector will clear. rc11 #77 — + // SCIM was previously bypassing this cleanup entirely, + // leaving orphan pm-tty-* and USER provision actions on + // every device the deleted user was assigned to. + user, loadErr := h.store.Queries().GetUserByID(ctx, userID) + err = h.store.AppendEvent(ctx, store.Event{ StreamType: "user", StreamID: userID, @@ -859,6 +867,21 @@ func (h *Handler) deleteUser(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusInternalServerError, "failed to delete user") return } + + // Best-effort cleanup. If the projection load above failed + // we have nothing to feed the cleaner; log and let the + // periodic reconciler eventually GC orphan actions via the + // is_deleted projection column. systemActions can be nil in + // tests. + if h.systemActions != nil && loadErr == nil { + if err := h.systemActions.CleanupDeletedUserActions(ctx, user); err != nil { + h.logger.Error("failed to cleanup system actions for SCIM-deleted user", + "user_id", userID, "error", err) + } + } else if loadErr != nil { + h.logger.Warn("could not load user projection for SCIM delete cleanup; orphan actions may remain", + "user_id", userID, "error", loadErr) + } } w.WriteHeader(http.StatusNoContent) diff --git a/internal/store/store.go b/internal/store/store.go index b7b2afb6..81c86e6b 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -7,6 +7,8 @@ import ( "encoding/json" "errors" "fmt" + "os" + "sync" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" @@ -32,16 +34,109 @@ type Event struct { ActorID string } +// EventListener is a post-commit hook fired after a successful +// AppendEvent. Listeners are invoked synchronously in registration +// order; a slow listener will stall subsequent listeners on the same +// event. None are allowed to mutate the event row. +// +// Context lifetime: the ctx passed in is the caller's request context +// and will be cancelled as soon as AppendEvent returns. Listeners that +// kick off async work must derive a fresh context (typically via +// context.WithoutCancel + context.WithTimeout) and not rely on the +// passed ctx for downstream DB calls. +// +// Panic isolation: panics from a listener are recovered by +// fireListeners and logged to stderr — the AppendEvent caller sees a +// successful commit even if a listener crashes. This matches the +// "post-commit notification" contract: the event is durable; listener +// failures are observability, not correctness. +type EventListener func(ctx context.Context, ev PersistedEvent) + // Store wraps the database connection and provides access to queries. type Store struct { pool *pgxpool.Pool queries *Queries - // OnEventAppended is called after every successful AppendEvent with the persisted row. - // Used to trigger search indexing without modifying each call site. + // listenersMu guards listeners + OnEventAppended. Documented + // usage is "register at boot, then start serving", but Go's race + // detector won't catch a future caller that registers after + // AppendEvent is in flight (concurrent slice append + range read + // is a data race). RWMutex is cheap on the hot read path + // (fireListeners holds RLock) and lets boot code register without + // extra ceremony. + listenersMu sync.RWMutex + + // listeners are invoked after every successful AppendEvent / + // AppendEventWithVersion. Used by search indexing and (rc11) by + // the system-action derived-projection reconciler. Direct field + // access deliberately replaced with RegisterEventListener so we + // can append additional consumers without callers stomping on + // each other. + listeners []EventListener + + // OnEventAppended is preserved for backwards compatibility with + // the search-indexing wiring at cmd/control/main.go. Setting it + // is equivalent to RegisterEventListener; do not read from it + // outside this file. Guarded by listenersMu — callers that + // reassign at runtime should go through RegisterEventListener + // instead. (rc11 review round 4: search-indexer wiring migrated.) OnEventAppended func(ctx context.Context, ev PersistedEvent) } +// RegisterEventListener appends a post-commit hook. Multiple +// listeners may be registered; they fire in registration order. +// Safe to call concurrently with AppendEvent — the RWMutex serialises +// against fireListeners' read iteration. +func (s *Store) RegisterEventListener(fn EventListener) { + if fn == nil { + return + } + s.listenersMu.Lock() + s.listeners = append(s.listeners, fn) + s.listenersMu.Unlock() +} + +// fireListeners invokes both the legacy OnEventAppended callback (if +// set) and every RegisterEventListener entry. Centralised so the two +// AppendEvent variants stay in sync. +// +// Each listener is wrapped in defer/recover so a panic in one cannot +// fail AppendEvent — the event is already committed, and listeners +// are notifications, not part of the write path. Round-3 review of +// rc11 #77 caught this: a panicking listener used to bubble up +// through AppendEvent and fail the RPC even though the event was +// durable, breaking the "event committed → RPC succeeds" contract. +// We log panics to stderr because the Store has no slog handle; the +// listener owner is responsible for richer logging inside its own +// body if needed. +func (s *Store) fireListeners(ctx context.Context, row PersistedEvent) { + safe := func(name string, fn func()) { + defer func() { + if r := recover(); r != nil { + fmt.Fprintf(os.Stderr, "store: %s listener panicked on %s/%s: %v\n", name, row.StreamType, row.EventType, r) + } + }() + fn() + } + // Snapshot under RLock so the dispatch loop runs without holding + // the mutex (listeners may take milliseconds; we don't want to + // block concurrent RegisterEventListener calls or other readers + // for that long). Slice header copy is safe because + // RegisterEventListener appends — never mutates an existing + // element. + s.listenersMu.RLock() + onEvent := s.OnEventAppended + listeners := s.listeners + s.listenersMu.RUnlock() + + if onEvent != nil { + safe("OnEventAppended", func() { onEvent(ctx, row) }) + } + for _, l := range listeners { + safe("RegisterEventListener", func() { l(ctx, row) }) + } +} + // New creates a new Store and runs migrations. // This should only be called by the control server which is responsible for database schema management. func New(ctx context.Context, connString string) (*Store, error) { @@ -188,9 +283,7 @@ func (s *Store) AppendEvent(ctx context.Context, event Event) error { } return fmt.Errorf("append event: %w", err) } - if s.OnEventAppended != nil { - s.OnEventAppended(ctx, row) - } + s.fireListeners(ctx, row) return nil } return fmt.Errorf("append event: exhausted retries") @@ -228,9 +321,7 @@ func (s *Store) AppendEventWithVersion(ctx context.Context, event Event, expecte } return fmt.Errorf("append event: %w", err) } - if s.OnEventAppended != nil { - s.OnEventAppended(ctx, row) - } + s.fireListeners(ctx, row) return nil }