diff --git a/cmd/pad/groups.go b/cmd/pad/groups.go index e834d8ca..a3efcc0b 100644 --- a/cmd/pad/groups.go +++ b/cmd/pad/groups.go @@ -49,6 +49,7 @@ func workspaceCmd() *cobra.Command { joinCmd(), exportCmd(), importCmd(), + auditLogCmd(), ) return cmd } @@ -143,6 +144,19 @@ func agentCmd() *cobra.Command { return cmd } +func dbCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "db", + Short: "Database backup, restore, and migration tools", + } + cmd.AddCommand( + dbBackupCmd(), + dbRestoreCmd(), + dbMigrateToPgCmd(), + ) + return cmd +} + func agentUpdateCmd() *cobra.Command { return &cobra.Command{ Use: "update", diff --git a/cmd/pad/main.go b/cmd/pad/main.go index 92c1f39d..be61b347 100644 --- a/cmd/pad/main.go +++ b/cmd/pad/main.go @@ -33,6 +33,7 @@ import ( "github.com/redis/go-redis/v9" "github.com/xarmian/pad/internal/events" "github.com/xarmian/pad/internal/logging" + "github.com/xarmian/pad/internal/metrics" "github.com/xarmian/pad/internal/models" "github.com/xarmian/pad/internal/server" "github.com/xarmian/pad/internal/store" @@ -86,6 +87,7 @@ func main() { githubCmd(), roleCmd(), webhooksCmd(), + dbCmd(), completionCmd(), ) @@ -223,6 +225,13 @@ func serveCmd() *cobra.Command { srv.SetBaseURL(cfg.BaseURL()) srv.SetCORSOrigins(cfg.CORSOrigins) srv.SetSecureCookies(cfg.SecureCookies) + srv.SetSSELimits(cfg.SSEMaxConnections, cfg.SSEMaxPerWorkspace) + + // Initialize Prometheus metrics + m := metrics.New() + m.RegisterDBCollector(s.DB()) + srv.SetMetrics(m) + slog.Info("Prometheus metrics enabled at /metrics") // Attach event bus for real-time SSE var eventBus events.EventBus @@ -241,6 +250,8 @@ func serveCmd() *cobra.Command { eventBus = events.New() slog.Info("Event bus using in-memory (single instance)") } + // Wrap event bus with Prometheus instrumentation + eventBus = metrics.NewInstrumentedBus(eventBus, m) srv.SetEventBus(eventBus) // Attach webhook dispatcher for outgoing notifications @@ -5190,3 +5201,342 @@ func relativeTimeStr(t time.Time) string { return fmt.Sprintf("%dd ago", int(d.Hours()/24)) } } + +// --- database tools --- + +// pgDbnameFromURL extracts just the database name from a PostgreSQL URL for display purposes. +func pgDbnameFromURL(raw string) string { + u, err := url.Parse(raw) + if err != nil { + return "unknown" + } + return strings.TrimPrefix(u.Path, "/") +} + +func dbBackupCmd() *cobra.Command { + var output string + var cronMode bool + + cmd := &cobra.Command{ + Use: "backup", + Short: "Back up the PostgreSQL database using pg_dump", + Long: `Creates a SQL dump of the Pad PostgreSQL database. +Requires pg_dump to be installed and PAD_DATABASE_URL or PAD_DB_DRIVER=postgres to be configured. + +For SQLite, simply copy the database file (default: ~/.pad/pad.db).`, + RunE: func(cmd *cobra.Command, args []string) error { + dbURL := os.Getenv("PAD_DATABASE_URL") + if dbURL == "" { + return fmt.Errorf("PAD_DATABASE_URL is not set. This command requires PostgreSQL.\nFor SQLite, copy the database file directly: cp ~/.pad/pad.db backup.db") + } + + if output == "" { + output = fmt.Sprintf("pad-backup-%s.sql", time.Now().Format("20060102-150405")) + } + + // Pass the connection URL via environment variable instead of + // command-line args, so credentials don't leak in ps/proc output. + // --clean emits DROP statements so the dump can be restored into an + // existing database, and --if-exists avoids errors on a fresh DB. + pgArgs := []string{ + "--format", "plain", + "--clean", + "--if-exists", + "--file", output, + } + + pgCmd := exec.Command("pg_dump", pgArgs...) + pgCmd.Env = append(os.Environ(), "PGDATABASE="+dbURL) + pgCmd.Stdout = os.Stdout + pgCmd.Stderr = os.Stderr + + dbname := pgDbnameFromURL(dbURL) + if !cronMode { + fmt.Fprintf(os.Stderr, "Backing up database %s to %s...\n", dbname, output) + } + + if err := pgCmd.Run(); err != nil { + if cronMode { + slog.Error("backup failed", "error", err, "output", output) + } + return fmt.Errorf("pg_dump failed: %w", err) + } + + // Get file size + if info, err := os.Stat(output); err == nil { + sizeMB := float64(info.Size()) / 1024 / 1024 + if cronMode { + slog.Info("backup completed", "output", output, "size_mb", fmt.Sprintf("%.1f", sizeMB)) + } else { + fmt.Fprintf(os.Stderr, "Backup complete: %s (%.1f MB)\n", output, sizeMB) + } + } + + return nil + }, + } + + cmd.Flags().StringVarP(&output, "output", "o", "", "output file path (default: pad-backup-YYYYMMDD-HHMMSS.sql)") + cmd.Flags().BoolVar(&cronMode, "cron", false, "cron mode: structured log output, no interactive messages") + + return cmd +} + +func dbRestoreCmd() *cobra.Command { + var force bool + + cmd := &cobra.Command{ + Use: "restore ", + Short: "Restore a PostgreSQL database from a backup", + Long: `Restores a Pad PostgreSQL database from a SQL dump created by 'pad db backup'. +Requires psql to be installed and PAD_DATABASE_URL to be configured. + +WARNING: This will overwrite the current database contents.`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + inputFile := args[0] + if _, err := os.Stat(inputFile); os.IsNotExist(err) { + return fmt.Errorf("backup file not found: %s", inputFile) + } + + dbURL := os.Getenv("PAD_DATABASE_URL") + if dbURL == "" { + return fmt.Errorf("PAD_DATABASE_URL is not set. This command requires PostgreSQL.") + } + + dbname := pgDbnameFromURL(dbURL) + if !force { + fmt.Fprintf(os.Stderr, "WARNING: This will overwrite the database '%s' with data from %s.\n", dbname, inputFile) + fmt.Fprintf(os.Stderr, "Run with --force to skip this confirmation, or press Ctrl+C to abort.\n") + fmt.Fprintf(os.Stderr, "Continue? [y/N] ") + var confirm string + fmt.Scanln(&confirm) + if confirm != "y" && confirm != "Y" { + fmt.Fprintln(os.Stderr, "Aborted.") + return nil + } + } + + // Pass the connection URL via environment variable instead of + // command-line args, so credentials don't leak in ps/proc output. + psqlArgs := []string{ + "--file", inputFile, + "--single-transaction", + } + + psqlCmd := exec.Command("psql", psqlArgs...) + psqlCmd.Env = append(os.Environ(), "PGDATABASE="+dbURL) + psqlCmd.Stdout = os.Stdout + psqlCmd.Stderr = os.Stderr + + fmt.Fprintf(os.Stderr, "Restoring database %s from %s...\n", dbname, inputFile) + + if err := psqlCmd.Run(); err != nil { + return fmt.Errorf("psql restore failed: %w", err) + } + + fmt.Fprintln(os.Stderr, "Restore complete.") + return nil + }, + } + + cmd.Flags().BoolVar(&force, "force", false, "skip confirmation prompt") + + return cmd +} + +func dbMigrateToPgCmd() *cobra.Command { + var fromPath string + var toURL string + + cmd := &cobra.Command{ + Use: "migrate-to-pg", + Short: "Migrate data from SQLite to PostgreSQL", + Long: `One-time migration from a SQLite database to PostgreSQL. +Uses application-level export/import to transfer all workspace data. + +This reads each workspace from the SQLite database and imports it into +the PostgreSQL database. Users, platform settings, and auth data are +NOT migrated — only workspace content (collections, items, comments, +links, versions). + +Steps: + 1. Set up a fresh PostgreSQL database + 2. Run 'pad serve' with PAD_DB_DRIVER=postgres once to create the schema + 3. Stop the server + 4. Run this command to migrate workspace data`, + RunE: func(cmd *cobra.Command, args []string) error { + if fromPath == "" { + fromPath = filepath.Join(os.Getenv("HOME"), ".pad", "pad.db") + } + if _, err := os.Stat(fromPath); os.IsNotExist(err) { + return fmt.Errorf("SQLite database not found: %s", fromPath) + } + + if toURL == "" { + toURL = os.Getenv("PAD_DATABASE_URL") + } + if toURL == "" { + return fmt.Errorf("target PostgreSQL URL required: use --to or set PAD_DATABASE_URL") + } + + // Open source SQLite + fmt.Fprintf(os.Stderr, "Opening SQLite database: %s\n", fromPath) + srcStore, err := store.New(fromPath) + if err != nil { + return fmt.Errorf("open SQLite: %w", err) + } + defer srcStore.Close() + + // Open target PostgreSQL + fmt.Fprintf(os.Stderr, "Connecting to PostgreSQL: %s\n", maskPassword(toURL)) + dstStore, err := store.NewPostgres(toURL) + if err != nil { + return fmt.Errorf("open PostgreSQL: %w", err) + } + defer dstStore.Close() + + // List workspaces from source + workspaces, err := srcStore.ListWorkspaces() + if err != nil { + return fmt.Errorf("list workspaces: %w", err) + } + + if len(workspaces) == 0 { + fmt.Fprintln(os.Stderr, "No workspaces found in SQLite database.") + return nil + } + + fmt.Fprintf(os.Stderr, "Found %d workspace(s) to migrate:\n", len(workspaces)) + for _, ws := range workspaces { + fmt.Fprintf(os.Stderr, " - %s (%s)\n", ws.Name, ws.Slug) + } + fmt.Fprintln(os.Stderr) + + migrated := 0 + for _, ws := range workspaces { + fmt.Fprintf(os.Stderr, "Migrating workspace: %s...\n", ws.Name) + + data, err := srcStore.ExportWorkspace(ws.Slug) + if err != nil { + fmt.Fprintf(os.Stderr, " ERROR exporting %s: %v (skipping)\n", ws.Slug, err) + continue + } + + stats := fmt.Sprintf("%d collections, %d items, %d comments", + len(data.Collections), len(data.Items), len(data.Comments)) + + if _, err := dstStore.ImportWorkspace(data, ""); err != nil { + fmt.Fprintf(os.Stderr, " ERROR importing %s: %v (skipping)\n", ws.Slug, err) + continue + } + + fmt.Fprintf(os.Stderr, " OK: %s\n", stats) + migrated++ + } + + fmt.Fprintf(os.Stderr, "\nMigration complete: %d/%d workspace(s) migrated.\n", migrated, len(workspaces)) + if migrated < len(workspaces) { + fmt.Fprintln(os.Stderr, "Some workspaces failed — check the errors above.") + return fmt.Errorf("%d workspace(s) failed to migrate", len(workspaces)-migrated) + } + + fmt.Fprintln(os.Stderr, "\nNext steps:") + fmt.Fprintln(os.Stderr, " 1. Set PAD_DB_DRIVER=postgres and PAD_DATABASE_URL in your environment") + fmt.Fprintln(os.Stderr, " 2. Start the server: pad serve") + fmt.Fprintln(os.Stderr, " 3. Run 'pad auth setup' to create an admin account on the new database") + fmt.Fprintln(os.Stderr, " 4. Verify your data in the web UI") + + return nil + }, + } + + cmd.Flags().StringVar(&fromPath, "from", "", "SQLite database path (default: ~/.pad/pad.db)") + cmd.Flags().StringVar(&toURL, "to", "", "PostgreSQL connection URL (default: PAD_DATABASE_URL)") + + return cmd +} + +// maskPassword replaces the password in a PostgreSQL URL for safe display. +func maskPassword(pgURL string) string { + u, err := url.Parse(pgURL) + if err != nil { + return "***" + } + if _, hasPW := u.User.Password(); hasPW { + u.User = url.UserPassword(u.User.Username(), "***") + } + return u.String() +} + +// --- audit-log --- + +func auditLogCmd() *cobra.Command { + var days int + var actor string + var action string + var limit int + + cmd := &cobra.Command{ + Use: "audit-log", + Short: "View the compliance audit log (admin-only)", + RunE: func(cmd *cobra.Command, args []string) error { + client, _ := getClient() + + params := models.AuditLogParams{ + Days: days, + Actor: actor, + Action: action, + Limit: limit, + } + + activities, err := client.GetAuditLog(params) + if err != nil { + return err + } + + if formatFlag == "json" { + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(activities) + } + + if len(activities) == 0 { + fmt.Println("No audit log entries found.") + return nil + } + + w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintln(w, "TIME\tACTION\tACTOR\tIP\tDETAILS") + for _, a := range activities { + ts := a.CreatedAt.Format("2006-01-02 15:04") + actorName := a.ActorName + if actorName == "" { + actorName = a.UserID + } + ip := a.IPAddress + if ip == "" { + ip = "-" + } + detail := a.Metadata + if detail == "" { + detail = "-" + } + // Truncate long metadata + if len(detail) > 60 { + detail = detail[:57] + "..." + } + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", ts, a.Action, actorName, ip, detail) + } + w.Flush() + return nil + }, + } + + cmd.Flags().IntVar(&days, "days", 30, "number of days to look back") + cmd.Flags().StringVar(&actor, "actor", "", "filter by actor (user ID)") + cmd.Flags().StringVar(&action, "action", "", "filter by action type") + cmd.Flags().IntVar(&limit, "limit", 50, "maximum number of entries") + + return cmd +} diff --git a/deploy/Caddyfile b/deploy/Caddyfile new file mode 100644 index 00000000..3b4b39d7 --- /dev/null +++ b/deploy/Caddyfile @@ -0,0 +1,31 @@ +# Pad — Caddy reverse proxy +# Caddy automatically provisions and renews TLS certificates. +# +# Usage: +# 1. Replace "pad.example.com" with your domain +# 2. Run: caddy run --config Caddyfile +# +# With Docker Compose, add a Caddy service on the pad-net network. + +pad.example.com { + reverse_proxy pad:7777 { + # SSE: disable buffering for real-time events + flush_interval -1 + } + + # Optional: rate limiting + # rate_limit { + # zone api { + # key {remote_host} + # events 100 + # window 1m + # } + # } + + encode gzip + + log { + output file /var/log/caddy/pad-access.log + format json + } +} diff --git a/deploy/k8s/configmap.yaml b/deploy/k8s/configmap.yaml new file mode 100644 index 00000000..d28c3f61 --- /dev/null +++ b/deploy/k8s/configmap.yaml @@ -0,0 +1,18 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: pad-config + namespace: pad +data: + PAD_HOST: "0.0.0.0" + PAD_PORT: "7777" + PAD_DB_DRIVER: "postgres" + PAD_DATA_DIR: "/data" + PAD_LOG_LEVEL: "info" + PAD_SECURE_COOKIES: "true" + PAD_SSE_MAX_CONNECTIONS: "1000" + PAD_SSE_MAX_PER_WORKSPACE: "100" + # Set your public URL: + # PAD_URL: "https://pad.example.com" + # CORS origins: + # PAD_CORS_ORIGINS: "https://pad.example.com" diff --git a/deploy/k8s/deployment.yaml b/deploy/k8s/deployment.yaml new file mode 100644 index 00000000..61a52a6a --- /dev/null +++ b/deploy/k8s/deployment.yaml @@ -0,0 +1,60 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: pad + namespace: pad + labels: + app: pad +spec: + replicas: 2 + selector: + matchLabels: + app: pad + template: + metadata: + labels: + app: pad + spec: + containers: + - name: pad + image: ghcr.io/xarmian/pad:latest + ports: + - containerPort: 7777 + name: http + envFrom: + - configMapRef: + name: pad-config + - secretRef: + name: pad-secret + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + cpu: "1" + memory: 512Mi + readinessProbe: + httpGet: + path: /api/v1/health + port: http + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 3 + failureThreshold: 3 + livenessProbe: + httpGet: + path: /api/v1/health + port: http + initialDelaySeconds: 15 + periodSeconds: 30 + timeoutSeconds: 5 + failureThreshold: 3 + volumeMounts: + - name: data + mountPath: /data + volumes: + - name: data + emptyDir: {} + # For SQLite mode, use a PersistentVolumeClaim instead: + # persistentVolumeClaim: + # claimName: pad-data diff --git a/deploy/k8s/hpa.yaml b/deploy/k8s/hpa.yaml new file mode 100644 index 00000000..df939d3d --- /dev/null +++ b/deploy/k8s/hpa.yaml @@ -0,0 +1,25 @@ +apiVersion: autoscaling/v2 +kind: HorizontalPodAutoscaler +metadata: + name: pad + namespace: pad +spec: + scaleTargetRef: + apiVersion: apps/v1 + kind: Deployment + name: pad + minReplicas: 2 + maxReplicas: 10 + metrics: + - type: Resource + resource: + name: cpu + target: + type: Utilization + averageUtilization: 70 + - type: Resource + resource: + name: memory + target: + type: Utilization + averageUtilization: 80 diff --git a/deploy/k8s/ingress.yaml b/deploy/k8s/ingress.yaml new file mode 100644 index 00000000..1c1880bd --- /dev/null +++ b/deploy/k8s/ingress.yaml @@ -0,0 +1,29 @@ +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: pad + namespace: pad + annotations: + # For nginx ingress controller: + nginx.ingress.kubernetes.io/proxy-buffering: "off" + nginx.ingress.kubernetes.io/proxy-read-timeout: "3600" + nginx.ingress.kubernetes.io/proxy-send-timeout: "3600" + # For cert-manager: + # cert-manager.io/cluster-issuer: "letsencrypt-prod" +spec: + ingressClassName: nginx + tls: + - hosts: + - pad.example.com + secretName: pad-tls + rules: + - host: pad.example.com + http: + paths: + - path: / + pathType: Prefix + backend: + service: + name: pad + port: + name: http diff --git a/deploy/k8s/namespace.yaml b/deploy/k8s/namespace.yaml new file mode 100644 index 00000000..1322c212 --- /dev/null +++ b/deploy/k8s/namespace.yaml @@ -0,0 +1,4 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: pad diff --git a/deploy/k8s/secret.yaml b/deploy/k8s/secret.yaml new file mode 100644 index 00000000..b3c174e7 --- /dev/null +++ b/deploy/k8s/secret.yaml @@ -0,0 +1,16 @@ +# IMPORTANT: Replace placeholder values before applying. +# Consider using sealed-secrets, external-secrets, or a vault integration +# instead of committing real credentials. +apiVersion: v1 +kind: Secret +metadata: + name: pad-secret + namespace: pad +type: Opaque +stringData: + PAD_DATABASE_URL: "postgres://pad:CHANGE_ME@postgres-host:5432/pad?sslmode=require" + PAD_REDIS_URL: "redis://:CHANGE_ME@redis-host:6379" + # Optional email: + # PAD_MAILEROO_API_KEY: "your-sending-key" + # PAD_EMAIL_FROM: "noreply@example.com" + # PAD_EMAIL_FROM_NAME: "Pad" diff --git a/deploy/k8s/service.yaml b/deploy/k8s/service.yaml new file mode 100644 index 00000000..5e82daa5 --- /dev/null +++ b/deploy/k8s/service.yaml @@ -0,0 +1,16 @@ +apiVersion: v1 +kind: Service +metadata: + name: pad + namespace: pad + labels: + app: pad +spec: + type: ClusterIP + ports: + - port: 80 + targetPort: http + protocol: TCP + name: http + selector: + app: pad diff --git a/deploy/nginx.conf b/deploy/nginx.conf new file mode 100644 index 00000000..5d55cf3b --- /dev/null +++ b/deploy/nginx.conf @@ -0,0 +1,83 @@ +# Pad — nginx reverse proxy +# +# Usage: +# 1. Replace "pad.example.com" with your domain +# 2. Update ssl_certificate paths to your TLS certs +# 3. Include this file in your nginx config or copy to /etc/nginx/conf.d/ +# +# Key settings for SSE support: +# - proxy_buffering off +# - proxy_read_timeout 86400s (24h for long-lived SSE connections) +# - proxy_http_version 1.1 with Connection "" + +upstream pad_backend { + server 127.0.0.1:7777; + # For Docker: server pad:7777; + keepalive 32; +} + +server { + listen 80; + server_name pad.example.com; + return 301 https://$host$request_uri; +} + +server { + listen 443 ssl http2; + server_name pad.example.com; + + ssl_certificate /etc/ssl/certs/pad.example.com.pem; + ssl_certificate_key /etc/ssl/private/pad.example.com-key.pem; + + # Modern TLS settings + ssl_protocols TLSv1.2 TLSv1.3; + ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384; + ssl_prefer_server_ciphers off; + + # Security headers + add_header X-Frame-Options "SAMEORIGIN" always; + add_header X-Content-Type-Options "nosniff" always; + add_header Referrer-Policy "strict-origin-when-cross-origin" always; + + # Gzip + gzip on; + gzip_types text/plain application/json text/css application/javascript; + + # SSE endpoint — requires special proxy settings + location /api/v1/events { + proxy_pass http://pad_backend; + proxy_http_version 1.1; + proxy_set_header Connection ""; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + # Critical for SSE: + proxy_buffering off; + proxy_cache off; + proxy_read_timeout 86400s; + proxy_send_timeout 86400s; + chunked_transfer_encoding on; + } + + # All other routes + location / { + proxy_pass http://pad_backend; + proxy_http_version 1.1; + proxy_set_header Connection ""; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + proxy_buffering off; + proxy_read_timeout 60s; + + # File uploads + client_max_body_size 10M; + } + + access_log /var/log/nginx/pad-access.log; + error_log /var/log/nginx/pad-error.log; +} diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml new file mode 100644 index 00000000..4afdf8b8 --- /dev/null +++ b/docker-compose.prod.yml @@ -0,0 +1,78 @@ +# Pad — production Docker Compose override +# Usage: docker compose -f docker-compose.yml -f docker-compose.prod.yml up -d +# +# Extends the base docker-compose.yml with: +# - Resource limits +# - Secure cookie settings +# - CORS configuration +# - Email (Maileroo) support +# - Named network for reverse proxy integration + +services: + pad: + environment: + PAD_SECURE_COOKIES: "true" + # Override Redis URL to include password when REDIS_PASSWORD is set. + # Without this, the pad container inherits the passwordless URL from + # docker-compose.yml and fails to connect when Redis AUTH is enabled. + PAD_REDIS_URL: "redis://:${REDIS_PASSWORD:-}@redis:6379" + # Set your public-facing URL for correct invitation links: + # PAD_URL: "https://pad.example.com" + # CORS origins (comma-separated): + # PAD_CORS_ORIGINS: "https://pad.example.com" + # Email (Maileroo): + # PAD_MAILEROO_API_KEY: "your-sending-key" + # PAD_EMAIL_FROM: "noreply@example.com" + # PAD_EMAIL_FROM_NAME: "Pad" + # SSE limits: + # PAD_SSE_MAX_CONNECTIONS: "1000" + # PAD_SSE_MAX_PER_WORKSPACE: "100" + deploy: + resources: + limits: + cpus: "2.0" + memory: 512M + reservations: + cpus: "0.5" + memory: 128M + networks: + - pad-net + - default + + postgres: + environment: + POSTGRES_PASSWORD: "${POSTGRES_PASSWORD:-change-me-in-production}" + deploy: + resources: + limits: + cpus: "2.0" + memory: 1G + reservations: + cpus: "0.25" + memory: 256M + # In production, consider using a managed PostgreSQL service instead. + + redis: + command: redis-server --maxmemory 128mb --maxmemory-policy allkeys-lru --requirepass "${REDIS_PASSWORD:-}" + healthcheck: + # Override the base healthcheck to authenticate when REDIS_PASSWORD is set. + # redis-cli reads REDISCLI_AUTH automatically for authentication. + test: ["CMD-SHELL", "REDISCLI_AUTH=$${REDIS_PASSWORD:-} redis-cli ping | grep -q PONG"] + interval: 5s + timeout: 3s + retries: 5 + environment: + REDIS_PASSWORD: "${REDIS_PASSWORD:-}" + deploy: + resources: + limits: + cpus: "1.0" + memory: 256M + reservations: + cpus: "0.1" + memory: 64M + +networks: + pad-net: + name: pad-net + # Attach your reverse proxy (Caddy, nginx) to this network. diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 00000000..dec6199c --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,68 @@ +# Pad — local production setup with PostgreSQL + Redis +# Usage: docker compose up -d +# +# Starts Pad with PostgreSQL for storage and Redis for real-time events. +# Access the web UI at http://localhost:7777 +# First-time setup: visit the UI or run `pad auth setup` from a local CLI. + +services: + pad: + build: + context: . + dockerfile: Dockerfile + ports: + - "7777:7777" + environment: + PAD_HOST: "0.0.0.0" + PAD_PORT: "7777" + PAD_DB_DRIVER: "postgres" + PAD_DATABASE_URL: "postgres://pad:pad@postgres:5432/pad?sslmode=disable" + PAD_REDIS_URL: "redis://redis:6379" + PAD_DATA_DIR: "/data" + PAD_LOG_LEVEL: "info" + volumes: + - pad-data:/data + depends_on: + postgres: + condition: service_healthy + redis: + condition: service_healthy + restart: unless-stopped + healthcheck: + test: ["CMD", "wget", "-q", "--spider", "http://localhost:7777/api/v1/health"] + interval: 10s + timeout: 5s + retries: 3 + start_period: 10s + + postgres: + image: postgres:17-alpine + environment: + POSTGRES_USER: pad + POSTGRES_PASSWORD: pad + POSTGRES_DB: pad + volumes: + - pg-data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U pad"] + interval: 5s + timeout: 3s + retries: 5 + restart: unless-stopped + + redis: + image: redis:7-alpine + command: redis-server --maxmemory 64mb --maxmemory-policy allkeys-lru + volumes: + - redis-data:/data + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 5 + restart: unless-stopped + +volumes: + pad-data: + pg-data: + redis-data: diff --git a/docs/backup.md b/docs/backup.md new file mode 100644 index 00000000..221ebd0a --- /dev/null +++ b/docs/backup.md @@ -0,0 +1,151 @@ +# Backup & Restore Guide + +## Overview + +Pad provides built-in tooling for database backup, restore, and migration between SQLite and PostgreSQL. + +| Command | Description | +|---------|-------------| +| `pad db backup` | PostgreSQL backup via `pg_dump` | +| `pad db restore ` | PostgreSQL restore via `psql` | +| `pad db migrate-to-pg` | One-time SQLite → PostgreSQL migration | +| `pad workspace export` | Application-level JSON export (portable) | +| `pad workspace import` | Application-level JSON import | + +## SQLite Backups + +SQLite stores everything in a single file (default: `~/.pad/pad.db`). Back it up by copying the file: + +```bash +# Simple file copy (stop the server first for consistency) +cp ~/.pad/pad.db ~/backups/pad-$(date +%Y%m%d).db + +# Or use SQLite's backup command (safe while server is running) +sqlite3 ~/.pad/pad.db ".backup ~/backups/pad-$(date +%Y%m%d).db" +``` + +## PostgreSQL Backups + +### Manual Backup + +```bash +# Create a SQL dump +pad db backup + +# Specify output file +pad db backup --output /backups/pad-backup.sql +``` + +Requires: +- `pg_dump` installed +- `PAD_DATABASE_URL` environment variable set + +### Automated Backups (Cron) + +```bash +# Add to crontab: daily backup at 2 AM +0 2 * * * PAD_DATABASE_URL="postgres://pad:secret@localhost:5432/pad" /usr/local/bin/pad db backup --cron --output /backups/pad-$(date +\%Y\%m\%d).sql +``` + +The `--cron` flag uses structured log output suitable for log aggregation systems. + +### Restore + +```bash +# Restore from backup (will prompt for confirmation) +pad db restore /backups/pad-backup.sql + +# Skip confirmation (for automated restore) +pad db restore --force /backups/pad-backup.sql +``` + +### Cloud Database Snapshots + +For managed PostgreSQL (AWS RDS, Google Cloud SQL, Azure Database): + +- **AWS RDS**: Use automated backups + manual snapshots via the AWS Console or CLI +- **Google Cloud SQL**: Enable automated backups in instance settings +- **Azure**: Configure automated backups via the portal + +These are generally preferred over `pg_dump` for large databases as they use filesystem-level snapshots. + +## Migrating SQLite → PostgreSQL + +When graduating from a local SQLite setup to production PostgreSQL: + +```bash +# 1. Set up PostgreSQL and create the database +createdb pad + +# 2. Run Pad once against PostgreSQL to create the schema +PAD_DB_DRIVER=postgres PAD_DATABASE_URL="postgres://pad:secret@localhost:5432/pad" pad serve & +# Wait a few seconds for migrations to run, then stop it +kill %1 + +# 3. Migrate workspace data +pad db migrate-to-pg \ + --from ~/.pad/pad.db \ + --to "postgres://pad:secret@localhost:5432/pad" + +# 4. Create an admin account on the new database +PAD_DB_DRIVER=postgres PAD_DATABASE_URL="postgres://pad:secret@localhost:5432/pad" pad auth setup + +# 5. Start the server with PostgreSQL +PAD_DB_DRIVER=postgres PAD_DATABASE_URL="postgres://pad:secret@localhost:5432/pad" pad serve +``` + +**What gets migrated:** +- Workspaces, collections, items, comments +- Item links (dependencies) +- Item versions (history) + +**What does NOT get migrated:** +- User accounts and sessions (re-create with `pad auth setup`) +- Platform settings (reconfigure in admin panel) +- Activity/audit log (starts fresh) + +## Application-Level Export/Import + +For portable workspace backups that work across SQLite and PostgreSQL: + +```bash +# Export a workspace to JSON +pad workspace export > my-workspace.json + +# Import into any Pad instance (SQLite or PostgreSQL) +pad workspace import < my-workspace.json + +# Import with a new name +pad workspace import --name "imported-workspace" < my-workspace.json +``` + +This format is database-agnostic and can be used to: +- Transfer workspaces between Pad instances +- Create workspace templates +- Back up individual workspaces + +## Backup Strategy Recommendations + +### Small Teams (SQLite) + +``` +Daily: Copy pad.db to a backup location +Weekly: Rotate old backups (keep 4 weeks) +``` + +### Production (PostgreSQL) + +``` +Continuous: WAL archiving (point-in-time recovery) +Daily: pg_dump via 'pad db backup --cron' +Weekly: Full filesystem snapshot (if using managed DB) +Monthly: Test restore procedure +``` + +### Disaster Recovery Checklist + +- [ ] Backups are being created on schedule +- [ ] Backups are stored off-site (different region/provider) +- [ ] Restore procedure has been tested recently +- [ ] Recovery time objective (RTO) is documented +- [ ] Recovery point objective (RPO) is documented diff --git a/docs/deployment.md b/docs/deployment.md new file mode 100644 index 00000000..22bced88 --- /dev/null +++ b/docs/deployment.md @@ -0,0 +1,246 @@ +# Deployment Guide + +Pad is a single Go binary with an embedded web UI. It supports SQLite (default) for single-node deployments and PostgreSQL + Redis for production multi-node setups. + +## Architecture + +``` + ┌─────────────────┐ + │ Reverse Proxy │ + │ (Caddy/nginx) │ + └────────┬────────┘ + │ :443 + ┌────────▼────────┐ + │ Pad │ + │ Go binary │ + │ (web UI + API) │ + └──┬──────────┬───┘ + │ │ + ┌────────▼──┐ ┌───▼────────┐ + │ PostgreSQL │ │ Redis │ + │ (storage) │ │ (pub/sub) │ + └────────────┘ └────────────┘ +``` + +- **Pad** serves the REST API and embedded SvelteKit web UI on a single port (default: 7777) +- **PostgreSQL** stores all data (workspaces, items, users, activity). SQLite works for single-node. +- **Redis** enables real-time SSE events across multiple Pad instances. Optional for single-node. + +## Quick Start with Docker Compose + +```bash +# Clone the repo +git clone https://github.com/xarmian/pad.git +cd pad + +# Start everything (Pad + PostgreSQL + Redis) +docker compose up -d + +# Check status +docker compose ps + +# View logs +docker compose logs -f pad +``` + +Access the web UI at **http://localhost:7777**. On first visit, you'll be prompted to create an admin account. + +### Production Docker Compose + +```bash +# Use the production overlay for resource limits and secure settings +docker compose -f docker-compose.yml -f docker-compose.prod.yml up -d +``` + +Edit `docker-compose.prod.yml` to set your domain, email credentials, and database password. + +## Environment Variables + +All configuration is via environment variables or a config file (`~/.pad/config.toml` / `/data/config.toml`). + +### Core + +| Variable | Default | Description | +|----------|---------|-------------| +| `PAD_HOST` | `127.0.0.1` | Listen address (`0.0.0.0` for Docker/production) | +| `PAD_PORT` | `7777` | Listen port | +| `PAD_URL` | — | Public-facing base URL (e.g., `https://pad.example.com`). Used for invitation links. | +| `PAD_DATA_DIR` | `~/.pad` | Data directory for SQLite DB, logs, and config | +| `PAD_LOG_LEVEL` | `info` | Log level: `debug`, `info`, `warn`, `error` | +| `PAD_MODE` | `local` | Mode: `local`, `remote`, `docker`, `cloud` | + +### Database + +| Variable | Default | Description | +|----------|---------|-------------| +| `PAD_DB_DRIVER` | `sqlite` | Database driver: `sqlite` or `postgres` | +| `PAD_DB_PATH` | `~/.pad/pad.db` | SQLite database path (ignored when using PostgreSQL) | +| `PAD_DATABASE_URL` | — | PostgreSQL connection string (required when `PAD_DB_DRIVER=postgres`) | + +### Real-time Events + +| Variable | Default | Description | +|----------|---------|-------------| +| `PAD_REDIS_URL` | — | Redis URL for cross-instance pub/sub. Without Redis, SSE events are in-process only. | +| `PAD_SSE_MAX_CONNECTIONS` | `1000` | Global maximum SSE connections | +| `PAD_SSE_MAX_PER_WORKSPACE` | `100` | Per-workspace maximum SSE connections | + +### Security + +| Variable | Default | Description | +|----------|---------|-------------| +| `PAD_SECURE_COOKIES` | `false` | Set `Secure` flag on session cookies (requires TLS) | +| `PAD_CORS_ORIGINS` | — | Comma-separated allowed CORS origins | + +### Email (Optional) + +Email enables sending workspace invitation links. Without it, users can still join via CLI invite codes. + +| Variable | Default | Description | +|----------|---------|-------------| +| `PAD_MAILEROO_API_KEY` | — | Maileroo sending API key | +| `PAD_EMAIL_FROM` | `noreply@getpad.dev` | Sender email address | +| `PAD_EMAIL_FROM_NAME` | `Pad` | Sender display name | + +## Deployment Options + +### Single Binary (SQLite) + +The simplest deployment — one binary, one file for the database. + +```bash +# Download or build +make build + +# Run directly +PAD_HOST=0.0.0.0 ./pad serve + +# Or install as a systemd service (see below) +``` + +Best for: single-user, small teams, evaluations. + +### Docker Compose (PostgreSQL + Redis) + +See [Quick Start](#quick-start-with-docker-compose) above. This is the recommended setup for teams. + +### Kubernetes + +Manifests are in `deploy/k8s/`. Apply them in order: + +```bash +# Create namespace +kubectl apply -f deploy/k8s/namespace.yaml + +# Configure secrets (edit first!) +kubectl apply -f deploy/k8s/secret.yaml + +# Deploy +kubectl apply -f deploy/k8s/configmap.yaml +kubectl apply -f deploy/k8s/deployment.yaml +kubectl apply -f deploy/k8s/service.yaml +kubectl apply -f deploy/k8s/ingress.yaml +kubectl apply -f deploy/k8s/hpa.yaml +``` + +**Prerequisites:** +- External PostgreSQL (e.g., AWS RDS, Cloud SQL, managed PG) +- External Redis (e.g., ElastiCache, Memorystore) +- Ingress controller (nginx-ingress or similar) +- TLS certificates (cert-manager recommended) + +### Systemd Service + +```ini +# /etc/systemd/system/pad.service +[Unit] +Description=Pad +After=network.target postgresql.service redis.service + +[Service] +Type=simple +User=pad +Group=pad +ExecStart=/usr/local/bin/pad serve +Environment=PAD_HOST=0.0.0.0 +Environment=PAD_DATA_DIR=/var/lib/pad +Environment=PAD_DB_DRIVER=postgres +Environment=PAD_DATABASE_URL=postgres://pad:secret@localhost:5432/pad +Environment=PAD_REDIS_URL=redis://localhost:6379 +Restart=always +RestartSec=5 + +[Install] +WantedBy=multi-user.target +``` + +```bash +sudo systemctl daemon-reload +sudo systemctl enable --now pad +``` + +## Reverse Proxy + +Pad needs a reverse proxy for TLS termination. SSE connections require specific proxy settings to avoid buffering. + +### Caddy (Recommended) + +Caddy handles TLS automatically. See `deploy/Caddyfile`: + +``` +pad.example.com { + reverse_proxy pad:7777 { + flush_interval -1 + } +} +``` + +### nginx + +See `deploy/nginx.conf`. Critical settings for SSE: + +```nginx +location /api/v1/events { + proxy_buffering off; + proxy_cache off; + proxy_read_timeout 86400s; + proxy_http_version 1.1; + proxy_set_header Connection ""; +} +``` + +## Monitoring + +Pad exposes Prometheus metrics at `/metrics` (unauthenticated). Key metrics: + +| Metric | Type | Description | +|--------|------|-------------| +| `pad_http_requests_total` | counter | Total HTTP requests by method, path, status | +| `pad_http_request_duration_seconds` | histogram | Request latency | +| `pad_http_response_size_bytes` | histogram | Response body sizes | +| `pad_sse_connections_active` | gauge | Current SSE connections | +| `pad_eventbus_publish_total` | counter | Events published | +| `pad_eventbus_subscribers` | gauge | Active event subscribers | +| `pad_db_open_connections` | gauge | Database connection pool stats | + +### Health Check + +```bash +curl http://localhost:7777/api/v1/health +# {"status":"ok"} +``` + +## Production Checklist + +- [ ] **Database:** PostgreSQL configured with `PAD_DB_DRIVER=postgres` +- [ ] **Redis:** Connected for multi-instance SSE (`PAD_REDIS_URL`) +- [ ] **TLS:** Reverse proxy with valid certificates +- [ ] **Secure cookies:** `PAD_SECURE_COOKIES=true` (requires TLS) +- [ ] **Public URL:** `PAD_URL` set to your public-facing domain +- [ ] **CORS:** `PAD_CORS_ORIGINS` set if serving from a different domain +- [ ] **Backups:** PostgreSQL backup strategy in place (see `docs/backup.md`) +- [ ] **Monitoring:** Prometheus scraping `/metrics` +- [ ] **Admin account:** Created via `pad auth setup` or web UI on first visit +- [ ] **Email (optional):** Maileroo configured for invitation emails +- [ ] **Resource limits:** Set in Docker Compose or K8s manifests +- [ ] **Log level:** `PAD_LOG_LEVEL=info` (use `debug` only for troubleshooting) diff --git a/go.mod b/go.mod index 8adf3cf0..f889f13f 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( ) require ( + github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dustin/go-humanize v1.0.1 // indirect @@ -28,13 +29,20 @@ require ( github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.20 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect + github.com/prometheus/client_golang v1.23.2 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.66.1 // indirect + github.com/prometheus/procfs v0.16.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/spf13/pflag v1.0.9 // indirect go.uber.org/atomic v1.11.0 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.42.0 // indirect golang.org/x/text v0.35.0 // indirect + google.golang.org/protobuf v1.36.8 // indirect modernc.org/libc v1.70.0 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect diff --git a/go.sum b/go.sum index e1387190..f5c66be1 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,7 @@ github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk= github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= @@ -45,10 +47,20 @@ github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHP github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= +github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs= github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= @@ -70,6 +82,8 @@ github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA= @@ -88,6 +102,8 @@ golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= +google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= +google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= diff --git a/internal/cli/client.go b/internal/cli/client.go index a8f7b192..440afbc2 100644 --- a/internal/cli/client.go +++ b/internal/cli/client.go @@ -461,6 +461,37 @@ func (c *Client) CheckSession() (*SessionResponse, error) { return &result, c.get("/auth/session", &result) } +// --- Audit Log --- + +// GetAuditLog fetches the global audit log (admin-only). +func (c *Client) GetAuditLog(params models.AuditLogParams) ([]models.Activity, error) { + q := url.Values{} + if params.Action != "" { + q.Set("action", params.Action) + } + if params.Actor != "" { + q.Set("actor", params.Actor) + } + if params.WorkspaceID != "" { + q.Set("workspace", params.WorkspaceID) + } + if params.Days > 0 { + q.Set("days", fmt.Sprintf("%d", params.Days)) + } + if params.Limit > 0 { + q.Set("limit", fmt.Sprintf("%d", params.Limit)) + } + if params.Offset > 0 { + q.Set("offset", fmt.Sprintf("%d", params.Offset)) + } + path := "/audit-log" + if qs := q.Encode(); qs != "" { + path += "?" + qs + } + var result []models.Activity + return result, c.get(path, &result) +} + // --- HTTP helpers --- type APIError struct { diff --git a/internal/config/config.go b/internal/config/config.go index 05662dc0..3edad1cc 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -40,19 +40,25 @@ type Config struct { // Security CORSOrigins string `toml:"cors_origins"` // Comma-separated allowed origins (e.g. "https://app.pad.dev,https://admin.pad.dev") SecureCookies bool `toml:"secure_cookies"` // Set Secure flag on cookies (requires TLS) + + // SSE limits + SSEMaxConnections int `toml:"sse_max_connections"` // Global max SSE connections (0 = unlimited) + SSEMaxPerWorkspace int `toml:"sse_max_per_workspace"` // Per-workspace max SSE connections (0 = unlimited) } func DefaultConfig() *Config { homeDir, _ := os.UserHomeDir() dataDir := filepath.Join(homeDir, ".pad") return &Config{ - Host: "127.0.0.1", - Port: 7777, - Editor: "", - LogLevel: "info", - DBPath: filepath.Join(dataDir, "pad.db"), - DataDir: dataDir, - ConfigPath: filepath.Join(dataDir, "config.toml"), + Host: "127.0.0.1", + Port: 7777, + Editor: "", + LogLevel: "info", + DBPath: filepath.Join(dataDir, "pad.db"), + DataDir: dataDir, + ConfigPath: filepath.Join(dataDir, "config.toml"), + SSEMaxConnections: 1000, + SSEMaxPerWorkspace: 100, } } @@ -128,6 +134,16 @@ func Load() (*Config, error) { if v := os.Getenv("PAD_SECURE_COOKIES"); v == "true" || v == "1" { cfg.SecureCookies = true } + if v := os.Getenv("PAD_SSE_MAX_CONNECTIONS"); v != "" { + if max, err := strconv.Atoi(v); err == nil { + cfg.SSEMaxConnections = max + } + } + if v := os.Getenv("PAD_SSE_MAX_PER_WORKSPACE"); v != "" { + if max, err := strconv.Atoi(v); err == nil { + cfg.SSEMaxPerWorkspace = max + } + } return cfg, nil } diff --git a/internal/events/bus.go b/internal/events/bus.go index 8aab1b87..fe056703 100644 --- a/internal/events/bus.go +++ b/internal/events/bus.go @@ -54,6 +54,12 @@ type EventBus interface { // Returns a buffered channel that will receive events for that workspace. Subscribe(workspaceID string) chan Event + // SubscribeIfAllowed atomically checks the global and per-workspace + // subscriber limits and, only if both are satisfied, subscribes in the + // same critical section. Returns (ch, true) on success or (nil, false) + // when a limit would be exceeded. Pass 0 for either limit to disable it. + SubscribeIfAllowed(workspaceID string, maxGlobal, maxPerWorkspace int) (chan Event, bool) + // Unsubscribe removes a subscriber and closes its channel. Unsubscribe(ch chan Event) @@ -65,6 +71,10 @@ type EventBus interface { // SubscriberCount returns the number of active local subscribers. SubscriberCount() int + + // WorkspaceSubscriberCount returns the number of active subscribers + // for a specific workspace. + WorkspaceSubscriberCount(workspaceID string) int } // subscriber wraps a channel with its workspace filter. @@ -101,6 +111,34 @@ func (b *MemoryBus) Subscribe(workspaceID string) chan Event { return ch } +// SubscribeIfAllowed atomically checks limits and subscribes. +func (b *MemoryBus) SubscribeIfAllowed(workspaceID string, maxGlobal, maxPerWorkspace int) (chan Event, bool) { + b.mu.Lock() + defer b.mu.Unlock() + + if maxGlobal > 0 && len(b.subscribers) >= maxGlobal { + return nil, false + } + if maxPerWorkspace > 0 { + count := 0 + for _, sub := range b.subscribers { + if sub.workspaceID == workspaceID { + count++ + } + } + if count >= maxPerWorkspace { + return nil, false + } + } + + ch := make(chan Event, 64) + b.subscribers[ch] = &subscriber{ + ch: ch, + workspaceID: workspaceID, + } + return ch, true +} + // Unsubscribe removes a subscriber and closes its channel. func (b *MemoryBus) Unsubscribe(ch chan Event) { b.mu.Lock() @@ -153,3 +191,16 @@ func (b *MemoryBus) SubscriberCount() int { defer b.mu.RUnlock() return len(b.subscribers) } + +// WorkspaceSubscriberCount returns the number of active subscribers for a workspace. +func (b *MemoryBus) WorkspaceSubscriberCount(workspaceID string) int { + b.mu.RLock() + defer b.mu.RUnlock() + count := 0 + for _, sub := range b.subscribers { + if sub.workspaceID == workspaceID { + count++ + } + } + return count +} diff --git a/internal/events/bus_test.go b/internal/events/bus_test.go index e2961983..7f19a7c7 100644 --- a/internal/events/bus_test.go +++ b/internal/events/bus_test.go @@ -230,6 +230,43 @@ func TestConcurrentAccess(t *testing.T) { } } +func TestWorkspaceSubscriberCount(t *testing.T) { + bus := New() + + // No subscribers initially + if got := bus.WorkspaceSubscriberCount("ws-1"); got != 0 { + t.Fatalf("expected 0, got %d", got) + } + + // Subscribe to ws-1 + ch1 := bus.Subscribe("ws-1") + ch2 := bus.Subscribe("ws-1") + ch3 := bus.Subscribe("ws-2") + + if got := bus.WorkspaceSubscriberCount("ws-1"); got != 2 { + t.Fatalf("expected 2 for ws-1, got %d", got) + } + if got := bus.WorkspaceSubscriberCount("ws-2"); got != 1 { + t.Fatalf("expected 1 for ws-2, got %d", got) + } + if got := bus.WorkspaceSubscriberCount("ws-3"); got != 0 { + t.Fatalf("expected 0 for ws-3, got %d", got) + } + + // Unsubscribe one from ws-1 + bus.Unsubscribe(ch1) + if got := bus.WorkspaceSubscriberCount("ws-1"); got != 1 { + t.Fatalf("expected 1 for ws-1 after unsubscribe, got %d", got) + } + + // Unsubscribe remaining + bus.Unsubscribe(ch2) + bus.Unsubscribe(ch3) + if got := bus.WorkspaceSubscriberCount("ws-1"); got != 0 { + t.Fatalf("expected 0 for ws-1 after all unsubscribed, got %d", got) + } +} + func TestPublishNoSubscribers(t *testing.T) { bus := New() // Should not panic diff --git a/internal/events/redis_bus.go b/internal/events/redis_bus.go index 552d4677..09ece4cc 100644 --- a/internal/events/redis_bus.go +++ b/internal/events/redis_bus.go @@ -77,6 +77,35 @@ func (b *RedisBus) Subscribe(workspaceID string) chan Event { return ch } +// SubscribeIfAllowed atomically checks limits and subscribes. +// NOTE: Limits are enforced against local (per-pod) subscriber counts only. +// In multi-replica deployments the effective cap is multiplied by the number +// of replicas. For truly global caps, use a Redis-backed counter. +func (b *RedisBus) SubscribeIfAllowed(workspaceID string, maxGlobal, maxPerWorkspace int) (chan Event, bool) { + b.mu.Lock() + defer b.mu.Unlock() + + if maxGlobal > 0 && len(b.subscribers) >= maxGlobal { + return nil, false + } + if maxPerWorkspace > 0 && b.wsCounts[workspaceID] >= maxPerWorkspace { + return nil, false + } + + ch := make(chan Event, 64) + b.subscribers[ch] = &subscriber{ + ch: ch, + workspaceID: workspaceID, + } + + b.wsCounts[workspaceID]++ + if b.wsCounts[workspaceID] == 1 { + b.startRedisSubscription(workspaceID) + } + + return ch, true +} + // Unsubscribe removes a local subscriber and closes its channel. // Cancels the Redis subscription if this was the last local subscriber for the workspace. func (b *RedisBus) Unsubscribe(ch chan Event) { @@ -143,6 +172,13 @@ func (b *RedisBus) SubscriberCount() int { return len(b.subscribers) } +// WorkspaceSubscriberCount returns the number of active local subscribers for a workspace. +func (b *RedisBus) WorkspaceSubscriberCount(workspaceID string) int { + b.mu.RLock() + defer b.mu.RUnlock() + return b.wsCounts[workspaceID] +} + // startRedisSubscription begins listening on a Redis channel for a workspace. // Must be called with b.mu held. func (b *RedisBus) startRedisSubscription(workspaceID string) { diff --git a/internal/metrics/instrumented_bus.go b/internal/metrics/instrumented_bus.go new file mode 100644 index 00000000..3c37cc55 --- /dev/null +++ b/internal/metrics/instrumented_bus.go @@ -0,0 +1,96 @@ +package metrics + +import ( + "sync" + + "github.com/xarmian/pad/internal/events" +) + +// InstrumentedBus wraps an events.EventBus to record Prometheus metrics +// for SSE connections (per workspace) and event publish counts. +// It implements the events.EventBus interface so it can be used as a +// drop-in replacement without changing the interface or its implementations. +type InstrumentedBus struct { + inner events.EventBus + metrics *Metrics + + mu sync.Mutex + workspaces map[chan events.Event]string // channel → workspaceID for gauge decrement +} + +// NewInstrumentedBus wraps an EventBus with Prometheus instrumentation. +func NewInstrumentedBus(inner events.EventBus, m *Metrics) *InstrumentedBus { + return &InstrumentedBus{ + inner: inner, + metrics: m, + workspaces: make(map[chan events.Event]string), + } +} + +// Subscribe delegates to the inner bus and increments the SSE connection gauge. +func (b *InstrumentedBus) Subscribe(workspaceID string) chan events.Event { + ch := b.inner.Subscribe(workspaceID) + + b.mu.Lock() + b.workspaces[ch] = workspaceID + b.mu.Unlock() + + (*b.metrics.SSEConnectionsActive).Inc() + (*b.metrics.EventBusSubscribers).Set(float64(b.inner.SubscriberCount())) + return ch +} + +// SubscribeIfAllowed delegates the atomic check-and-subscribe to the inner bus +// and updates Prometheus gauges on success. +func (b *InstrumentedBus) SubscribeIfAllowed(workspaceID string, maxGlobal, maxPerWorkspace int) (chan events.Event, bool) { + ch, ok := b.inner.SubscribeIfAllowed(workspaceID, maxGlobal, maxPerWorkspace) + if !ok { + return nil, false + } + + b.mu.Lock() + b.workspaces[ch] = workspaceID + b.mu.Unlock() + + (*b.metrics.SSEConnectionsActive).Inc() + (*b.metrics.EventBusSubscribers).Set(float64(b.inner.SubscriberCount())) + return ch, true +} + +// Unsubscribe delegates to the inner bus and decrements the SSE connection gauge. +func (b *InstrumentedBus) Unsubscribe(ch chan events.Event) { + b.mu.Lock() + _, ok := b.workspaces[ch] + if ok { + delete(b.workspaces, ch) + } + b.mu.Unlock() + + b.inner.Unsubscribe(ch) + + if ok { + (*b.metrics.SSEConnectionsActive).Dec() + } + (*b.metrics.EventBusSubscribers).Set(float64(b.inner.SubscriberCount())) +} + +// Publish delegates to the inner bus and increments the publish counter. +func (b *InstrumentedBus) Publish(event events.Event) { + b.inner.Publish(event) + (*b.metrics.EventBusPublishTotal).Inc() +} + +// Close delegates to the inner bus. +func (b *InstrumentedBus) Close() { + b.inner.Close() +} + +// SubscriberCount delegates to the inner bus. +func (b *InstrumentedBus) SubscriberCount() int { + return b.inner.SubscriberCount() +} + +// WorkspaceSubscriberCount delegates to the inner bus. +func (b *InstrumentedBus) WorkspaceSubscriberCount(workspaceID string) int { + return b.inner.WorkspaceSubscriberCount(workspaceID) +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 00000000..4065d394 --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,147 @@ +// Package metrics provides Prometheus instrumentation for the Pad server. +// It uses a custom registry (not the global default) for test isolation +// and explicit control over exposed metrics. +package metrics + +import ( + "database/sql" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" +) + +// Metrics holds all Prometheus collectors and the custom registry. +type Metrics struct { + Registry *prometheus.Registry + + // HTTP request metrics + HTTPRequestsTotal *prometheus.CounterVec + HTTPRequestDuration *prometheus.HistogramVec + HTTPResponseSize *prometheus.HistogramVec + + // SSE connection metrics (single gauge to avoid unbounded label cardinality) + SSEConnectionsActive *prometheus.Gauge + + // EventBus metrics + EventBusPublishTotal *prometheus.Counter + EventBusSubscribers *prometheus.Gauge +} + +// New creates a new Metrics instance with a custom registry and registers +// all application metrics plus Go runtime and process collectors. +func New() *Metrics { + reg := prometheus.NewRegistry() + + // Go runtime + process collectors (goroutines, memory, GC, file descriptors) + reg.MustRegister(collectors.NewGoCollector()) + reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})) + + httpRequestsTotal := prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "pad_http_requests_total", + Help: "Total number of HTTP requests by method, route, and status code.", + }, []string{"method", "route", "status"}) + + httpRequestDuration := prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "pad_http_request_duration_seconds", + Help: "HTTP request duration in seconds by method, route, and status code.", + Buckets: prometheus.DefBuckets, + }, []string{"method", "route", "status"}) + + httpResponseSize := prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "pad_http_response_size_bytes", + Help: "HTTP response size in bytes by method, route, and status code.", + Buckets: prometheus.ExponentialBuckets(100, 10, 7), // 100B to 100MB + }, []string{"method", "route", "status"}) + + sseConnectionsActive := prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "pad_sse_connections_active", + Help: "Total number of active SSE connections.", + }) + + eventBusPublishTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "pad_eventbus_publish_total", + Help: "Total number of events published to the event bus.", + }) + + eventBusSubscribers := prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "pad_eventbus_subscribers", + Help: "Current number of event bus subscribers.", + }) + + reg.MustRegister( + httpRequestsTotal, + httpRequestDuration, + httpResponseSize, + sseConnectionsActive, + eventBusPublishTotal, + eventBusSubscribers, + ) + + return &Metrics{ + Registry: reg, + HTTPRequestsTotal: httpRequestsTotal, + HTTPRequestDuration: httpRequestDuration, + HTTPResponseSize: httpResponseSize, + SSEConnectionsActive: &sseConnectionsActive, + EventBusPublishTotal: &eventBusPublishTotal, + EventBusSubscribers: &eventBusSubscribers, + } +} + +// RegisterDBCollector registers a callback-based collector that exposes +// database connection pool statistics on each Prometheus scrape. +// This is preferred over a periodic goroutine: zero overhead between +// scrapes and always fresh data. +func (m *Metrics) RegisterDBCollector(db *sql.DB) { + m.Registry.MustRegister(&dbStatsCollector{db: db}) +} + +// dbStatsCollector implements prometheus.Collector using db.Stats() callbacks. +type dbStatsCollector struct { + db *sql.DB +} + +var ( + dbOpenDesc = prometheus.NewDesc( + "pad_db_open_connections", + "Number of open database connections.", + nil, nil, + ) + dbIdleDesc = prometheus.NewDesc( + "pad_db_idle_connections", + "Number of idle database connections.", + nil, nil, + ) + dbInUseDesc = prometheus.NewDesc( + "pad_db_in_use_connections", + "Number of in-use database connections.", + nil, nil, + ) + dbWaitCountDesc = prometheus.NewDesc( + "pad_db_wait_count_total", + "Total number of connections waited for.", + nil, nil, + ) + dbWaitDurationDesc = prometheus.NewDesc( + "pad_db_wait_duration_seconds_total", + "Total time blocked waiting for a new connection.", + nil, nil, + ) +) + +func (c *dbStatsCollector) Describe(ch chan<- *prometheus.Desc) { + ch <- dbOpenDesc + ch <- dbIdleDesc + ch <- dbInUseDesc + ch <- dbWaitCountDesc + ch <- dbWaitDurationDesc +} + +func (c *dbStatsCollector) Collect(ch chan<- prometheus.Metric) { + stats := c.db.Stats() + ch <- prometheus.MustNewConstMetric(dbOpenDesc, prometheus.GaugeValue, float64(stats.OpenConnections)) + ch <- prometheus.MustNewConstMetric(dbIdleDesc, prometheus.GaugeValue, float64(stats.Idle)) + ch <- prometheus.MustNewConstMetric(dbInUseDesc, prometheus.GaugeValue, float64(stats.InUse)) + ch <- prometheus.MustNewConstMetric(dbWaitCountDesc, prometheus.CounterValue, float64(stats.WaitCount)) + ch <- prometheus.MustNewConstMetric(dbWaitDurationDesc, prometheus.CounterValue, stats.WaitDuration.Seconds()) +} diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go new file mode 100644 index 00000000..649c1370 --- /dev/null +++ b/internal/metrics/metrics_test.go @@ -0,0 +1,191 @@ +package metrics + +import ( + "database/sql" + "strings" + "testing" + + "github.com/prometheus/client_golang/prometheus" + io_prometheus_client "github.com/prometheus/client_model/go" + + "github.com/xarmian/pad/internal/events" + + _ "modernc.org/sqlite" +) + +func TestNew(t *testing.T) { + m := New() + if m.Registry == nil { + t.Fatal("Registry should not be nil") + } + if m.HTTPRequestsTotal == nil { + t.Fatal("HTTPRequestsTotal should not be nil") + } + if m.HTTPRequestDuration == nil { + t.Fatal("HTTPRequestDuration should not be nil") + } + if m.HTTPResponseSize == nil { + t.Fatal("HTTPResponseSize should not be nil") + } + if m.SSEConnectionsActive == nil { + t.Fatal("SSEConnectionsActive should not be nil") + } + if m.EventBusPublishTotal == nil { + t.Fatal("EventBusPublishTotal should not be nil") + } + if m.EventBusSubscribers == nil { + t.Fatal("EventBusSubscribers should not be nil") + } + + // Verify all metrics can be gathered without error + families, err := m.Registry.Gather() + if err != nil { + t.Fatalf("Gather failed: %v", err) + } + + // Should have Go runtime + process metrics + found := false + for _, f := range families { + if strings.HasPrefix(f.GetName(), "go_") { + found = true + break + } + } + if !found { + t.Fatal("Expected Go runtime metrics to be registered") + } +} + +func TestRegisterDBCollector(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatalf("Failed to open in-memory SQLite: %v", err) + } + defer db.Close() + + m := New() + m.RegisterDBCollector(db) + + families, err := m.Registry.Gather() + if err != nil { + t.Fatalf("Gather failed: %v", err) + } + + expected := map[string]bool{ + "pad_db_open_connections": false, + "pad_db_idle_connections": false, + "pad_db_in_use_connections": false, + "pad_db_wait_count_total": false, + "pad_db_wait_duration_seconds_total": false, + } + + for _, f := range families { + if _, ok := expected[f.GetName()]; ok { + expected[f.GetName()] = true + } + } + + for name, found := range expected { + if !found { + t.Errorf("Expected metric %q not found in gathered metrics", name) + } + } +} + +func TestInstrumentedBus_SubscribeUnsubscribe(t *testing.T) { + inner := events.New() + m := New() + bus := NewInstrumentedBus(inner, m) + + // Subscribe to a workspace + ch := bus.Subscribe("ws-1") + if ch == nil { + t.Fatal("Subscribe should return a channel") + } + + // Check SSE gauge incremented (single total gauge, not per-workspace) + gauge := getGaugeValue(t, *m.SSEConnectionsActive) + if gauge != 1 { + t.Errorf("Expected SSE active connections = 1, got %v", gauge) + } + + // Subscribe a second connection to same workspace + ch2 := bus.Subscribe("ws-1") + gauge = getGaugeValue(t, *m.SSEConnectionsActive) + if gauge != 2 { + t.Errorf("Expected SSE active connections = 2, got %v", gauge) + } + + // Subscribe to a different workspace + ch3 := bus.Subscribe("ws-2") + gauge = getGaugeValue(t, *m.SSEConnectionsActive) + if gauge != 3 { + t.Errorf("Expected SSE active connections = 3, got %v", gauge) + } + + // Unsubscribe one from ws-1 + bus.Unsubscribe(ch) + gauge = getGaugeValue(t, *m.SSEConnectionsActive) + if gauge != 2 { + t.Errorf("Expected SSE active connections = 2 after unsubscribe, got %v", gauge) + } + + // Unsubscribe remaining + bus.Unsubscribe(ch2) + bus.Unsubscribe(ch3) + gauge = getGaugeValue(t, *m.SSEConnectionsActive) + if gauge != 0 { + t.Errorf("Expected SSE active connections = 0 after all unsubscribed, got %v", gauge) + } +} + +func TestInstrumentedBus_Publish(t *testing.T) { + inner := events.New() + m := New() + bus := NewInstrumentedBus(inner, m) + + // Subscribe so we can publish + ch := bus.Subscribe("ws-1") + defer bus.Unsubscribe(ch) + + bus.Publish(events.Event{Type: "test", WorkspaceID: "ws-1"}) + bus.Publish(events.Event{Type: "test", WorkspaceID: "ws-1"}) + + var metric io_prometheus_client.Metric + if err := (*m.EventBusPublishTotal).Write(&metric); err != nil { + t.Fatalf("Failed to write publish metric: %v", err) + } + if got := metric.GetCounter().GetValue(); got != 2 { + t.Errorf("Expected publish count = 2, got %v", got) + } +} + +func TestInstrumentedBus_SubscriberCount(t *testing.T) { + inner := events.New() + m := New() + bus := NewInstrumentedBus(inner, m) + + if bus.SubscriberCount() != 0 { + t.Errorf("Expected 0 subscribers initially") + } + + ch := bus.Subscribe("ws-1") + if bus.SubscriberCount() != 1 { + t.Errorf("Expected 1 subscriber after subscribe") + } + + bus.Unsubscribe(ch) + if bus.SubscriberCount() != 0 { + t.Errorf("Expected 0 subscribers after unsubscribe") + } +} + +// getGaugeValue extracts the current value from a Prometheus Gauge. +func getGaugeValue(t *testing.T, g prometheus.Gauge) float64 { + t.Helper() + var metric io_prometheus_client.Metric + if err := g.Write(&metric); err != nil { + t.Fatalf("Failed to write gauge metric: %v", err) + } + return metric.GetGauge().GetValue() +} diff --git a/internal/models/activity.go b/internal/models/activity.go index 2729a41a..cd7af44a 100644 --- a/internal/models/activity.go +++ b/internal/models/activity.go @@ -2,20 +2,39 @@ package models import "time" -// Valid actions +// Item-level actions (existing) var ValidActions = []string{ "created", "updated", "archived", "restored", "moved", "read", "searched", } +// Audit action constants for auth/admin events +const ( + ActionLogin = "login" + ActionLoginFailed = "login_failed" + ActionLogout = "logout" + ActionBootstrap = "bootstrap" + ActionRegister = "register" + ActionPasswordChanged = "password_changed" + ActionPasswordReset = "password_reset" + ActionTokenCreated = "token_created" + ActionTokenRevoked = "token_revoked" + ActionMemberInvited = "member_invited" + ActionMemberRemoved = "member_removed" + ActionRoleChanged = "role_changed" + ActionSettingsChanged = "settings_changed" +) + type Activity struct { ID string `json:"id"` - WorkspaceID string `json:"workspace_id"` + WorkspaceID string `json:"workspace_id,omitempty"` DocumentID string `json:"document_id,omitempty"` Action string `json:"action"` Actor string `json:"actor"` Source string `json:"source"` Metadata string `json:"metadata,omitempty"` // JSON UserID string `json:"user_id,omitempty"` + IPAddress string `json:"ip_address,omitempty"` + UserAgent string `json:"user_agent,omitempty"` CreatedAt time.Time `json:"created_at"` // Enrichment fields — populated by handlers, not stored in DB @@ -33,6 +52,16 @@ type ActivityListParams struct { Offset int } +// AuditLogParams are query parameters for the audit log endpoint. +type AuditLogParams struct { + Action string + Actor string + WorkspaceID string + Days int + Limit int + Offset int +} + // TimelineEntry represents a single entry in the unified item timeline. // It wraps one of: a comment, an activity, or a version. type TimelineEntry struct { diff --git a/internal/server/handlers_admin.go b/internal/server/handlers_admin.go index 6a33b88d..d462b637 100644 --- a/internal/server/handlers_admin.go +++ b/internal/server/handlers_admin.go @@ -1,7 +1,11 @@ package server import ( + "encoding/json" + "fmt" "net/http" + + "github.com/xarmian/pad/internal/models" ) // Known platform setting keys. Values are stored in the platform_settings table. @@ -75,6 +79,16 @@ func (s *Server) handleUpdatePlatformSettings(w http.ResponseWriter, r *http.Req // Reconfigure email sender if email settings changed s.reconfigureEmail() + // Log which settings were changed (keys only, not values for security) + var keys []string + for key := range input { + if allowed[key] { + keys = append(keys, key) + } + } + keysJSON, _ := json.Marshal(keys) + s.logAuditEvent(models.ActionSettingsChanged, r, fmt.Sprintf(`{"keys":%s}`, keysJSON)) + writeJSON(w, http.StatusOK, map[string]interface{}{"ok": true}) } diff --git a/internal/server/handlers_audit.go b/internal/server/handlers_audit.go new file mode 100644 index 00000000..b074da3a --- /dev/null +++ b/internal/server/handlers_audit.go @@ -0,0 +1,55 @@ +package server + +import ( + "net/http" + "strconv" + + "github.com/xarmian/pad/internal/models" +) + +// handleAuditLog returns a filtered audit log. Admin-only. +// Supports filtering by action, user (user ID), workspace, days, and pagination. +func (s *Server) handleAuditLog(w http.ResponseWriter, r *http.Request) { + user := currentUser(r) + if user == nil || user.Role != "admin" { + writeError(w, http.StatusForbidden, "forbidden", "Admin access required") + return + } + + // Accept both "user" and "actor" query params for filtering by user ID. + actorFilter := r.URL.Query().Get("user") + if actorFilter == "" { + actorFilter = r.URL.Query().Get("actor") + } + + params := models.AuditLogParams{ + Action: r.URL.Query().Get("action"), + Actor: actorFilter, + WorkspaceID: r.URL.Query().Get("workspace"), + Days: 30, // default + } + + if d := r.URL.Query().Get("days"); d != "" { + if days, err := strconv.Atoi(d); err == nil && days > 0 { + params.Days = days + } + } + if l := r.URL.Query().Get("limit"); l != "" { + if limit, err := strconv.Atoi(l); err == nil && limit > 0 { + params.Limit = limit + } + } + if o := r.URL.Query().Get("offset"); o != "" { + if offset, err := strconv.Atoi(o); err == nil && offset >= 0 { + params.Offset = offset + } + } + + activities, err := s.store.ListAuditLog(params) + if err != nil { + writeInternalError(w, err) + return + } + + writeJSON(w, http.StatusOK, activities) +} diff --git a/internal/server/handlers_auth.go b/internal/server/handlers_auth.go index f7ea280d..fee394c9 100644 --- a/internal/server/handlers_auth.go +++ b/internal/server/handlers_auth.go @@ -162,6 +162,8 @@ func (s *Server) handleBootstrap(w http.ResponseWriter, r *http.Request) { return } + s.logAuditEventForUser(models.ActionBootstrap, r, user.ID, auditMeta(map[string]string{"email": user.Email})) + writeJSON(w, http.StatusCreated, map[string]interface{}{ "user": sessionUserPayload(user), "token": token, @@ -274,6 +276,8 @@ func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) { return } + s.logAuditEventForUser(models.ActionRegister, r, user.ID, auditMeta(map[string]string{"email": user.Email})) + writeJSON(w, http.StatusCreated, map[string]interface{}{ "user": sessionUserPayload(user), "token": token, @@ -310,6 +314,7 @@ func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) { if user == nil { // Slow down brute force attempts time.Sleep(500 * time.Millisecond) + s.logAuditEvent(models.ActionLoginFailed, r, auditMeta(map[string]string{"email": input.Email})) writeError(w, http.StatusUnauthorized, "unauthorized", "Invalid email or password") return } @@ -319,6 +324,8 @@ func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) { return } + s.logAuditEventForUser(models.ActionLogin, r, user.ID, auditMeta(map[string]string{"email": user.Email})) + writeJSON(w, http.StatusOK, map[string]interface{}{ "user": sessionUserPayload(user), "token": token, @@ -388,6 +395,8 @@ func (s *Server) handleLogout(w http.ResponseWriter, r *http.Request) { // Clear CSRF cookie on logout clearCSRFCookie(w) + s.logAuditEvent(models.ActionLogout, r, "") + writeJSON(w, http.StatusOK, map[string]interface{}{ "ok": true, }) @@ -492,6 +501,10 @@ func (s *Server) handleUpdateCurrentUser(w http.ResponseWriter, r *http.Request) return } + if input.NewPassword != "" { + s.logAuditEvent(models.ActionPasswordChanged, r, "") + } + writeJSON(w, http.StatusOK, map[string]interface{}{ "id": updated.ID, "email": updated.Email, @@ -621,6 +634,8 @@ func (s *Server) handleResetPassword(w http.ResponseWriter, r *http.Request) { // Set CSRF cookie alongside the new session setCSRFCookie(w, int(webSessionTTL.Seconds()), s.secureCookies) + s.logAuditEventForUser(models.ActionPasswordReset, r, user.ID, auditMeta(map[string]string{"email": user.Email})) + writeJSON(w, http.StatusOK, map[string]interface{}{ "ok": true, "user": map[string]interface{}{ diff --git a/internal/server/handlers_documents.go b/internal/server/handlers_documents.go index 8f5d0a0c..f2b9052c 100644 --- a/internal/server/handlers_documents.go +++ b/internal/server/handlers_documents.go @@ -414,11 +414,12 @@ func agentMeta(r *http.Request, existingMeta string) string { return existingMeta } if existingMeta == "" || existingMeta == "{}" { - return fmt.Sprintf(`{"agent":"%s"}`, agentName) + return auditMeta(map[string]string{"agent": agentName}) } - // Merge: insert agent field into existing JSON + // Merge: insert agent field into existing JSON object if strings.HasPrefix(existingMeta, "{") { - return fmt.Sprintf(`{"agent":"%s",%s`, agentName, existingMeta[1:]) + agentJSON, _ := json.Marshal(agentName) + return fmt.Sprintf(`{"agent":%s,%s`, agentJSON, existingMeta[1:]) } return existingMeta } @@ -448,6 +449,62 @@ func (s *Server) logActivityWithMetaReturningID(workspaceID, documentID, action Source: source, Metadata: metadata, UserID: currentUserID(r), + IPAddress: clientIP(r), + UserAgent: r.Header.Get("User-Agent"), + }) +} + +// logAuditEvent logs a non-workspace audit event (e.g. login, logout). +// Best-effort: errors are silently ignored. +func (s *Server) logAuditEvent(action string, r *http.Request, metadata string) { + s.logAuditEventForUser(action, r, currentUserID(r), metadata) +} + +// logAuditEventForUser logs an audit event with an explicit user ID. +// Use this when the user isn't (yet) in the request context, e.g. after +// a successful login/register/bootstrap where the session was just created. +func (s *Server) logAuditEventForUser(action string, r *http.Request, userID string, metadata string) { + actor, source := actorFromRequest(r) + if metadata == "" { + metadata = "{}" + } + _, _ = s.store.CreateActivity(models.Activity{ + Action: action, + Actor: actor, + Source: source, + Metadata: metadata, + UserID: userID, + IPAddress: clientIP(r), + UserAgent: r.Header.Get("User-Agent"), + }) +} + +// auditMeta safely marshals a map to a JSON string for audit log metadata. +// Falls back to "{}" on marshal error so audit calls never break. +func auditMeta(kv map[string]string) string { + data, err := json.Marshal(kv) + if err != nil { + return "{}" + } + return string(data) +} + +// logWorkspaceAuditEvent logs a workspace-scoped audit event (e.g. member invited). +// Best-effort: errors are silently ignored. +func (s *Server) logWorkspaceAuditEvent(workspaceID, action string, r *http.Request, metadata string) { + actor, source := actorFromRequest(r) + if metadata == "" { + metadata = "{}" + } + _, _ = s.store.CreateActivity(models.Activity{ + WorkspaceID: workspaceID, + Action: action, + Actor: actor, + Source: source, + Metadata: metadata, + UserID: currentUserID(r), + IPAddress: clientIP(r), + UserAgent: r.Header.Get("User-Agent"), }) } diff --git a/internal/server/handlers_events.go b/internal/server/handlers_events.go index 3a1152e2..681ac14c 100644 --- a/internal/server/handlers_events.go +++ b/internal/server/handlers_events.go @@ -47,10 +47,27 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { w.Header().Set("Connection", "keep-alive") w.Header().Set("X-Accel-Buffering", "no") // Disable nginx buffering - // Subscribe to events for this workspace - ch := s.events.Subscribe(ws.ID) + // Atomically check SSE connection limits and subscribe in one step. + // This prevents TOCTOU races where two concurrent requests both pass the + // limit check before either subscribes. + ch, ok := s.events.SubscribeIfAllowed(ws.ID, s.sseMaxConnections, s.sseMaxPerWorkspace) + if !ok { + slog.Warn("SSE connection limit reached", "workspace", ws.Slug, + "global_current", s.events.SubscriberCount(), "global_max", s.sseMaxConnections, + "ws_current", s.events.WorkspaceSubscriberCount(ws.ID), "ws_max", s.sseMaxPerWorkspace) + writeError(w, http.StatusTooManyRequests, "sse_limit_exceeded", "SSE connection limit reached") + return + } defer s.events.Unsubscribe(ch) + // Log warning at 80% global capacity + if s.sseMaxConnections > 0 { + total := s.events.SubscriberCount() + if total >= s.sseMaxConnections*80/100 { + slog.Warn("SSE connections approaching global limit", "current", total, "max", s.sseMaxConnections) + } + } + // Send initial connected event writeSSEEvent(w, "connected", map[string]string{ "workspace_id": ws.ID, diff --git a/internal/server/handlers_events_test.go b/internal/server/handlers_events_test.go index 50e221b9..58aa21cf 100644 --- a/internal/server/handlers_events_test.go +++ b/internal/server/handlers_events_test.go @@ -360,3 +360,105 @@ func TestSSENoEventBus(t *testing.T) { t.Errorf("expected 503, got %d", rr.Code) } } + +func TestSSEGlobalConnectionLimit(t *testing.T) { + srv := testServerWithEvents(t) + srv.SetSSELimits(1, 0) // global limit of 1, no per-workspace limit + ts := httptest.NewServer(srv) + defer ts.Close() + + slug := createTestWorkspace(t, ts.URL, "Test") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // First connection should succeed + ch := connectSSE(ctx, t, ts.URL, slug) + waitForEvent(t, ch, 3*time.Second) // connected + + // Second connection should be rejected with 429 + req, err := http.NewRequest("GET", ts.URL+"/api/v1/events?workspace="+slug, nil) + if err != nil { + t.Fatal(err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusTooManyRequests { + t.Errorf("expected 429, got %d", resp.StatusCode) + } +} + +func TestSSEPerWorkspaceLimit(t *testing.T) { + srv := testServerWithEvents(t) + srv.SetSSELimits(0, 1) // no global limit, per-workspace limit of 1 + ts := httptest.NewServer(srv) + defer ts.Close() + + slug1 := createTestWorkspace(t, ts.URL, "WS One") + slug2 := createTestWorkspace(t, ts.URL, "WS Two") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // First connection to ws1 should succeed + ch1 := connectSSE(ctx, t, ts.URL, slug1) + waitForEvent(t, ch1, 3*time.Second) // connected + + // Second connection to ws1 should be rejected + req, err := http.NewRequest("GET", ts.URL+"/api/v1/events?workspace="+slug1, nil) + if err != nil { + t.Fatal(err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusTooManyRequests { + t.Errorf("expected 429 for second ws1 connection, got %d", resp.StatusCode) + } + + // Connection to ws2 should still succeed (different workspace) + ch2 := connectSSE(ctx, t, ts.URL, slug2) + event := waitForEvent(t, ch2, 3*time.Second) + if event.Type != "connected" { + t.Errorf("expected 'connected' for ws2, got %q", event.Type) + } +} + +func TestSSELimitsExistingConnectionsUnaffected(t *testing.T) { + srv := testServerWithEvents(t) + srv.SetSSELimits(1, 0) // global limit of 1 + ts := httptest.NewServer(srv) + defer ts.Close() + + slug := createTestWorkspace(t, ts.URL, "Test") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Establish connection + ch := connectSSE(ctx, t, ts.URL, slug) + waitForEvent(t, ch, 3*time.Second) // connected + + // Try (and fail) to get a second connection + req, _ := http.NewRequest("GET", ts.URL+"/api/v1/events?workspace="+slug, nil) + resp, _ := http.DefaultClient.Do(req) + resp.Body.Close() + + // The existing connection should still work — publish an event + srv.events.Publish(events.Event{ + Type: "item.created", + WorkspaceID: slug, // need the real workspace ID + }) + + // We can't easily test the existing connection receives events here + // because the workspace ID in the event must match the internal UUID, + // but we can verify the subscriber count is still 1 + if got := srv.events.SubscriberCount(); got != 1 { + t.Errorf("expected 1 subscriber still active, got %d", got) + } +} diff --git a/internal/server/handlers_items.go b/internal/server/handlers_items.go index 1e5b29f1..e55b355e 100644 --- a/internal/server/handlers_items.go +++ b/internal/server/handlers_items.go @@ -338,7 +338,8 @@ func (s *Server) handleUpdateItem(w http.ResponseWriter, r *http.Request) { } if input.Title != nil && *input.Title != item.Title { if meta == "" { - meta = fmt.Sprintf(`{"changes":"title: %s → %s"}`, item.Title, *input.Title) + titleChange := fmt.Sprintf("title: %s → %s", item.Title, *input.Title) + meta = fmt.Sprintf(`{"changes":%q}`, titleChange) } } // Track role and assignment changes @@ -566,7 +567,7 @@ func (s *Server) handleMoveItem(w http.ResponseWriter, r *http.Request) { // Log activity with metadata about the move actor, source := actorFromRequest(r) - moveMeta := fmt.Sprintf(`{"from_collection":"%s","to_collection":"%s"}`, sourceColl.Slug, targetColl.Slug) + moveMeta := auditMeta(map[string]string{"from_collection": sourceColl.Slug, "to_collection": targetColl.Slug}) s.logActivityWithMeta(workspaceID, moved.ID, "moved", r, moveMeta) // Publish events for both old and new collections diff --git a/internal/server/handlers_members.go b/internal/server/handlers_members.go index 9b5209c0..ee818406 100644 --- a/internal/server/handlers_members.go +++ b/internal/server/handlers_members.go @@ -6,6 +6,8 @@ import ( "net/http" "github.com/go-chi/chi/v5" + + "github.com/xarmian/pad/internal/models" ) // handleListMembers returns all members of a workspace. @@ -106,6 +108,7 @@ func (s *Server) handleInviteMember(w http.ResponseWriter, r *http.Request) { writeInternalError(w, err) return } + s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberInvited, r, auditMeta(map[string]string{"email": existingUser.Email, "role": input.Role, "added_directly": "true"})) writeJSON(w, http.StatusCreated, map[string]interface{}{ "added": true, "user_id": existingUser.ID, @@ -135,6 +138,8 @@ func (s *Server) handleInviteMember(w http.ResponseWriter, r *http.Request) { resp["join_url"] = joinURL } + s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberInvited, r, auditMeta(map[string]string{"email": input.Email, "role": input.Role})) + writeJSON(w, http.StatusCreated, resp) // Send invitation email asynchronously (fire-and-forget) @@ -180,6 +185,8 @@ func (s *Server) handleRemoveMember(w http.ResponseWriter, r *http.Request) { return } + s.logWorkspaceAuditEvent(workspaceID, models.ActionMemberRemoved, r, auditMeta(map[string]string{"user_id": userID})) + w.WriteHeader(http.StatusNoContent) } @@ -215,6 +222,8 @@ func (s *Server) handleUpdateMemberRole(w http.ResponseWriter, r *http.Request) return } + s.logWorkspaceAuditEvent(workspaceID, models.ActionRoleChanged, r, auditMeta(map[string]string{"user_id": userID, "role": input.Role})) + writeJSON(w, http.StatusOK, map[string]interface{}{ "user_id": userID, "role": input.Role, diff --git a/internal/server/handlers_tokens.go b/internal/server/handlers_tokens.go index 2e3b38c0..fc6edfbc 100644 --- a/internal/server/handlers_tokens.go +++ b/internal/server/handlers_tokens.go @@ -37,6 +37,8 @@ func (s *Server) handleCreateToken(w http.ResponseWriter, r *http.Request) { return } + s.logAuditEvent(models.ActionTokenCreated, r, auditMeta(map[string]string{"name": input.Name, "workspace_id": input.WorkspaceID})) + writeJSON(w, http.StatusCreated, token) } @@ -76,6 +78,8 @@ func (s *Server) handleDeleteToken(w http.ResponseWriter, r *http.Request) { return } + s.logAuditEvent(models.ActionTokenRevoked, r, auditMeta(map[string]string{"token_id": tokenID})) + w.WriteHeader(http.StatusNoContent) } @@ -126,6 +130,8 @@ func (s *Server) handleCreateUserToken(w http.ResponseWriter, r *http.Request) { return } + s.logAuditEvent(models.ActionTokenCreated, r, auditMeta(map[string]string{"name": input.Name})) + writeJSON(w, http.StatusCreated, token) } @@ -147,5 +153,7 @@ func (s *Server) handleDeleteUserToken(w http.ResponseWriter, r *http.Request) { return } + s.logAuditEvent(models.ActionTokenRevoked, r, auditMeta(map[string]string{"token_id": tokenID})) + w.WriteHeader(http.StatusNoContent) } diff --git a/internal/server/middleware_metrics.go b/internal/server/middleware_metrics.go new file mode 100644 index 00000000..802424f6 --- /dev/null +++ b/internal/server/middleware_metrics.go @@ -0,0 +1,40 @@ +package server + +import ( + "net/http" + "strconv" + "time" + + "github.com/go-chi/chi/v5" + chimiddleware "github.com/go-chi/chi/v5/middleware" + + "github.com/xarmian/pad/internal/metrics" +) + +// MetricsMiddleware returns a chi middleware that records Prometheus metrics +// for every HTTP request: count, duration, and response size. +// +// Path labels use chi's RoutePattern (e.g. "/api/v1/workspaces/{slug}/items/{itemSlug}") +// instead of actual URL paths, keeping label cardinality bounded. +func MetricsMiddleware(m *metrics.Metrics) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + ww := chimiddleware.NewWrapResponseWriter(w, r.ProtoMajor) + + next.ServeHTTP(ww, r) + + duration := time.Since(start).Seconds() + status := strconv.Itoa(ww.Status()) + route := chi.RouteContext(r.Context()).RoutePattern() + if route == "" { + route = "unmatched" + } + method := r.Method + + m.HTTPRequestsTotal.WithLabelValues(method, route, status).Inc() + m.HTTPRequestDuration.WithLabelValues(method, route, status).Observe(duration) + m.HTTPResponseSize.WithLabelValues(method, route, status).Observe(float64(ww.BytesWritten())) + }) + } +} diff --git a/internal/server/middleware_metrics_test.go b/internal/server/middleware_metrics_test.go new file mode 100644 index 00000000..90b91944 --- /dev/null +++ b/internal/server/middleware_metrics_test.go @@ -0,0 +1,134 @@ +package server + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/go-chi/chi/v5" + io_prometheus_client "github.com/prometheus/client_model/go" + + "github.com/xarmian/pad/internal/metrics" +) + +func TestMetricsMiddleware_RecordsRequestMetrics(t *testing.T) { + m := metrics.New() + + r := chi.NewRouter() + r.Use(MetricsMiddleware(m)) + r.Get("/api/v1/items/{slug}", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"ok":true}`)) + }) + + req := httptest.NewRequest("GET", "/api/v1/items/my-item", nil) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("Expected 200, got %d", rec.Code) + } + + // Check request count + counter, err := m.HTTPRequestsTotal.GetMetricWithLabelValues("GET", "/api/v1/items/{slug}", "200") + if err != nil { + t.Fatalf("Failed to get counter: %v", err) + } + var metric io_prometheus_client.Metric + if err := counter.Write(&metric); err != nil { + t.Fatalf("Failed to write counter: %v", err) + } + if got := metric.GetCounter().GetValue(); got != 1 { + t.Errorf("Expected request count = 1, got %v", got) + } + + // Check duration and response size via the registry gather + families, gatherErr := m.Registry.Gather() + if gatherErr != nil { + t.Fatalf("Failed to gather metrics: %v", gatherErr) + } + + var durationCount uint64 + var responseSizeSum float64 + for _, f := range families { + switch f.GetName() { + case "pad_http_request_duration_seconds": + for _, fm := range f.GetMetric() { + durationCount += fm.GetHistogram().GetSampleCount() + } + case "pad_http_response_size_bytes": + for _, fm := range f.GetMetric() { + responseSizeSum += fm.GetHistogram().GetSampleSum() + } + } + } + + if durationCount != 1 { + t.Errorf("Expected 1 observation in duration histogram, got %d", durationCount) + } + if responseSizeSum <= 0 { + t.Errorf("Expected response size > 0, got %v", responseSizeSum) + } +} + +func TestMetricsMiddleware_UsesRoutePattern(t *testing.T) { + m := metrics.New() + + r := chi.NewRouter() + r.Use(MetricsMiddleware(m)) + r.Get("/api/v1/workspaces/{ws}/items/{slug}", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + + // Hit with different slugs — should all map to the same route pattern + for _, path := range []string{ + "/api/v1/workspaces/my-ws/items/item-1", + "/api/v1/workspaces/other-ws/items/item-2", + "/api/v1/workspaces/third/items/item-3", + } { + req := httptest.NewRequest("GET", path, nil) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + } + + // All 3 requests should be under the same route pattern label + counter, err := m.HTTPRequestsTotal.GetMetricWithLabelValues("GET", "/api/v1/workspaces/{ws}/items/{slug}", "200") + if err != nil { + t.Fatalf("Failed to get counter: %v", err) + } + var metric io_prometheus_client.Metric + if err := counter.Write(&metric); err != nil { + t.Fatalf("Failed to write counter: %v", err) + } + if got := metric.GetCounter().GetValue(); got != 3 { + t.Errorf("Expected 3 requests under route pattern, got %v", got) + } +} + +func TestMetricsMiddleware_UnmatchedRoute(t *testing.T) { + m := metrics.New() + + r := chi.NewRouter() + r.Use(MetricsMiddleware(m)) + r.Get("/api/v1/health", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + + // Hit a route that doesn't exist + req := httptest.NewRequest("GET", "/nonexistent", nil) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + + // chi returns 404 for unmatched routes; the middleware should label it "unmatched" + counter, err := m.HTTPRequestsTotal.GetMetricWithLabelValues("GET", "unmatched", "404") + if err != nil { + t.Fatalf("Failed to get counter: %v", err) + } + var metric io_prometheus_client.Metric + if err := counter.Write(&metric); err != nil { + t.Fatalf("Failed to write counter: %v", err) + } + if got := metric.GetCounter().GetValue(); got != 1 { + t.Errorf("Expected 1 unmatched request, got %v", got) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 26d42d16..3a37e692 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -14,9 +14,11 @@ import ( "github.com/go-chi/chi/v5" chimiddleware "github.com/go-chi/chi/v5/middleware" "github.com/go-chi/cors" + "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/xarmian/pad/internal/email" "github.com/xarmian/pad/internal/events" + "github.com/xarmian/pad/internal/metrics" "github.com/xarmian/pad/internal/models" "github.com/xarmian/pad/internal/store" "github.com/xarmian/pad/internal/webhooks" @@ -35,7 +37,10 @@ type Server struct { baseURL string // public base URL for generating links (e.g. invite URLs) corsOrigins string // comma-separated CORS origins (empty = localhost defaults) secureCookies bool // set Secure flag on cookies (for TLS deployments) - version string // release version (e.g. "dev", "1.2.3") + metrics *metrics.Metrics // Prometheus metrics (optional) + sseMaxConnections int // global SSE connection limit (0 = unlimited) + sseMaxPerWorkspace int // per-workspace SSE connection limit (0 = unlimited) + version string // release version (e.g. "dev", "1.2.3") commit string // git commit hash buildTime string // build timestamp } @@ -84,6 +89,19 @@ func (s *Server) SetSecureCookies(secure bool) { s.secureCookies = secure } +// SetMetrics attaches Prometheus metrics to the server. +// Must be called before the first request is served. +func (s *Server) SetMetrics(m *metrics.Metrics) { + s.metrics = m +} + +// SetSSELimits configures global and per-workspace SSE connection limits. +// A value of 0 means unlimited. +func (s *Server) SetSSELimits(global, perWorkspace int) { + s.sseMaxConnections = global + s.sseMaxPerWorkspace = perWorkspace +} + // reconfigureEmail reads email settings from the platform_settings table // and updates (or creates) the email sender. Called after admin settings change. func (s *Server) reconfigureEmail() { @@ -113,34 +131,47 @@ func (s *Server) InitEmailFromSettings() { func (s *Server) setupRouter() { r := chi.NewRouter() - // Middleware + // Infrastructure middleware (applies to all routes including /metrics) r.Use(chimiddleware.RealIP) r.Use(chimiddleware.RequestID) r.Use(StructuredLogger) r.Use(chimiddleware.Recoverer) - r.Use(SecurityHeaders) - if s.secureCookies { - r.Use(StrictTransportSecurity) + if s.metrics != nil { + r.Use(MetricsMiddleware(s.metrics)) } - r.Use(cors.Handler(cors.Options{ - AllowedOrigins: parseCORSOrigins(s.corsOrigins), - AllowedMethods: []string{"GET", "POST", "PATCH", "PUT", "DELETE", "OPTIONS"}, - AllowedHeaders: []string{"Accept", "Authorization", "Content-Type", "X-CSRF-Token"}, - AllowCredentials: true, - MaxAge: 300, - })) - r.Use(s.TokenAuth) - r.Use(s.SessionAuth) - r.Use(s.RateLimit) - r.Use(s.CSRFProtect) - r.Use(s.RequireAuth) - r.Use(jsonContentType) - // SSE endpoint (outside jsonContentType middleware) - r.Get("/api/v1/events", s.handleSSE) + // Prometheus scrape endpoint — no auth/CSRF/security headers + if s.metrics != nil { + r.Group(func(r chi.Router) { + r.Handle("/metrics", promhttp.HandlerFor(s.metrics.Registry, promhttp.HandlerOpts{})) + }) + } - // API routes - r.Route("/api/v1", func(r chi.Router) { + // All other routes — full middleware stack + r.Group(func(r chi.Router) { + r.Use(SecurityHeaders) + if s.secureCookies { + r.Use(StrictTransportSecurity) + } + r.Use(cors.Handler(cors.Options{ + AllowedOrigins: parseCORSOrigins(s.corsOrigins), + AllowedMethods: []string{"GET", "POST", "PATCH", "PUT", "DELETE", "OPTIONS"}, + AllowedHeaders: []string{"Accept", "Authorization", "Content-Type", "X-CSRF-Token"}, + AllowCredentials: true, + MaxAge: 300, + })) + r.Use(s.TokenAuth) + r.Use(s.SessionAuth) + r.Use(s.RateLimit) + r.Use(s.CSRFProtect) + r.Use(s.RequireAuth) + r.Use(jsonContentType) + + // SSE endpoint (outside jsonContentType middleware — but inherits auth) + r.Get("/api/v1/events", s.handleSSE) + + // API routes + r.Route("/api/v1", func(r chi.Router) { r.Get("/health", s.handleHealth) r.Get("/health/live", s.handleHealthLive) r.Get("/health/ready", s.handleHealthReady) @@ -172,6 +203,9 @@ func (s *Server) setupRouter() { r.Post("/test-email", s.handleTestEmail) }) + // Audit log (admin-only) + r.Get("/audit-log", s.handleAuditLog) + // Templates r.Get("/templates", s.handleListTemplates) @@ -325,6 +359,7 @@ func (s *Server) setupRouter() { // Search r.Get("/search", s.handleSearch) }) + }) // end r.Group (full middleware stack) s.router = r } diff --git a/internal/store/activities.go b/internal/store/activities.go index 40c127de..eb7bf68a 100644 --- a/internal/store/activities.go +++ b/internal/store/activities.go @@ -21,9 +21,9 @@ func (s *Store) CreateActivity(a models.Activity) (string, error) { ts := now() _, err := s.db.Exec(s.q(` - INSERT INTO activities (id, workspace_id, document_id, action, actor, source, metadata, user_id, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - `), a.ID, a.WorkspaceID, nilIfEmpty(a.DocumentID), a.Action, a.Actor, a.Source, a.Metadata, nilIfEmpty(a.UserID), ts) + INSERT INTO activities (id, workspace_id, document_id, action, actor, source, metadata, user_id, ip_address, user_agent, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `), a.ID, nilIfEmpty(a.WorkspaceID), nilIfEmpty(a.DocumentID), a.Action, a.Actor, a.Source, a.Metadata, nilIfEmpty(a.UserID), nilIfEmpty(a.IPAddress), nilIfEmpty(a.UserAgent), ts) return a.ID, err } @@ -118,7 +118,7 @@ func mergeActivityMeta(existing, incoming string) string { func (s *Store) ListWorkspaceActivity(workspaceID string, params models.ActivityListParams) ([]models.Activity, error) { query := ` - SELECT a.id, a.workspace_id, COALESCE(a.document_id, ''), a.action, a.actor, a.source, a.metadata, COALESCE(a.user_id, ''), a.created_at, COALESCE(u.name, '') + SELECT a.id, COALESCE(a.workspace_id, ''), COALESCE(a.document_id, ''), a.action, a.actor, a.source, a.metadata, COALESCE(a.user_id, ''), a.created_at, COALESCE(u.name, ''), COALESCE(a.ip_address, ''), COALESCE(a.user_agent, '') FROM activities a LEFT JOIN users u ON a.user_id = u.id WHERE a.workspace_id = ? @@ -160,7 +160,7 @@ func (s *Store) ListWorkspaceActivity(workspaceID string, params models.Activity func (s *Store) ListDocumentActivity(documentID string, params models.ActivityListParams) ([]models.Activity, error) { query := ` - SELECT a.id, a.workspace_id, COALESCE(a.document_id, ''), a.action, a.actor, a.source, a.metadata, COALESCE(a.user_id, ''), a.created_at, COALESCE(u.name, '') + SELECT a.id, COALESCE(a.workspace_id, ''), COALESCE(a.document_id, ''), a.action, a.actor, a.source, a.metadata, COALESCE(a.user_id, ''), a.created_at, COALESCE(u.name, ''), COALESCE(a.ip_address, ''), COALESCE(a.user_agent, '') FROM activities a LEFT JOIN users u ON a.user_id = u.id WHERE a.document_id = ? @@ -201,7 +201,7 @@ func (s *Store) ListDocumentActivity(documentID string, params models.ActivityLi func (s *Store) ListDocumentActivityBeforeTime(documentID string, before time.Time, beforeID string, limit int) ([]models.Activity, error) { ts := before.Format(time.RFC3339) rows, err := s.db.Query(s.q(` - SELECT a.id, a.workspace_id, COALESCE(a.document_id, ''), a.action, a.actor, a.source, a.metadata, COALESCE(a.user_id, ''), a.created_at, COALESCE(u.name, '') + SELECT a.id, COALESCE(a.workspace_id, ''), COALESCE(a.document_id, ''), a.action, a.actor, a.source, a.metadata, COALESCE(a.user_id, ''), a.created_at, COALESCE(u.name, ''), COALESCE(a.ip_address, ''), COALESCE(a.user_agent, '') FROM activities a LEFT JOIN users u ON a.user_id = u.id WHERE a.document_id = ? AND (a.created_at < ? OR (a.created_at = ? AND a.id < ?)) @@ -225,7 +225,7 @@ func scanActivitiesWithUser(rows interface { for rows.Next() { var a models.Activity var createdAt string - if err := rows.Scan(&a.ID, &a.WorkspaceID, &a.DocumentID, &a.Action, &a.Actor, &a.Source, &a.Metadata, &a.UserID, &createdAt, &a.ActorName); err != nil { + if err := rows.Scan(&a.ID, &a.WorkspaceID, &a.DocumentID, &a.Action, &a.Actor, &a.Source, &a.Metadata, &a.UserID, &createdAt, &a.ActorName, &a.IPAddress, &a.UserAgent); err != nil { return nil, err } a.CreatedAt = parseTime(createdAt) @@ -240,3 +240,57 @@ func nilIfEmpty(s string) interface{} { } return s } + +// ListAuditLog returns activities matching the given audit log filters. +// Supports filtering by action, actor, workspace, and date range. +func (s *Store) ListAuditLog(params models.AuditLogParams) ([]models.Activity, error) { + // Build the full query with ? placeholders first, then rebind once at + // the end so PostgreSQL $1/$2/... numbering is correct across all filters. + query := ` + SELECT a.id, COALESCE(a.workspace_id, ''), COALESCE(a.document_id, ''), a.action, a.actor, a.source, a.metadata, COALESCE(a.user_id, ''), a.created_at, COALESCE(u.name, ''), COALESCE(a.ip_address, ''), COALESCE(a.user_agent, '') + FROM activities a + LEFT JOIN users u ON a.user_id = u.id + WHERE 1=1 + ` + args := []interface{}{} + + if params.WorkspaceID != "" { + query += ` AND a.workspace_id = ?` + args = append(args, params.WorkspaceID) + } + if params.Action != "" { + query += ` AND a.action = ?` + args = append(args, params.Action) + } + if params.Actor != "" { + query += ` AND a.user_id = ?` + args = append(args, params.Actor) + } + if params.Days > 0 { + cutoff := time.Now().UTC().AddDate(0, 0, -params.Days).Format(time.RFC3339) + query += ` AND a.created_at >= ?` + args = append(args, cutoff) + } + + query += ` ORDER BY a.created_at DESC` + + limit := params.Limit + if limit <= 0 { + limit = 50 + } + query += fmt.Sprintf(` LIMIT %d`, limit) + if params.Offset > 0 { + query += fmt.Sprintf(` OFFSET %d`, params.Offset) + } + + // Rebind all ? placeholders in one pass so $1, $2, ... are sequential. + query = s.q(query) + + rows, err := s.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + return scanActivitiesWithUser(rows) +} diff --git a/internal/store/migrations/022_audit_trail.sql b/internal/store/migrations/022_audit_trail.sql new file mode 100644 index 00000000..c57fcd24 --- /dev/null +++ b/internal/store/migrations/022_audit_trail.sql @@ -0,0 +1,40 @@ +-- Audit trail: extend activities with IP/UA, nullable workspace, expanded action types. +-- SQLite requires table recreation to alter CHECK constraints. + +PRAGMA foreign_keys = OFF; + +DROP TABLE IF EXISTS activities_new; + +CREATE TABLE activities_new ( + id TEXT PRIMARY KEY, + workspace_id TEXT REFERENCES workspaces(id), + document_id TEXT, + action TEXT NOT NULL, + actor TEXT NOT NULL, + source TEXT NOT NULL DEFAULT 'web', + metadata TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + user_id TEXT REFERENCES users(id), + ip_address TEXT, + user_agent TEXT +); + +-- Copy existing data (ip_address and user_agent will be NULL for historical records) +INSERT INTO activities_new (id, workspace_id, document_id, action, actor, source, metadata, created_at, user_id) +SELECT id, workspace_id, document_id, action, actor, source, metadata, created_at, user_id +FROM activities; + +-- Update foreign keys pointing to activities +-- comments.activity_id references activities(id) — handled automatically by SQLite +-- since we're using the same primary key values + +DROP TABLE activities; +ALTER TABLE activities_new RENAME TO activities; + +-- Recreate indexes +CREATE INDEX IF NOT EXISTS idx_activities_workspace ON activities(workspace_id, created_at); +CREATE INDEX IF NOT EXISTS idx_activities_document ON activities(document_id, created_at); +CREATE INDEX IF NOT EXISTS idx_activities_action ON activities(action, created_at); +CREATE INDEX IF NOT EXISTS idx_activities_user ON activities(user_id, created_at); + +PRAGMA foreign_keys = ON; diff --git a/internal/store/pgmigrations/002_audit_trail.sql b/internal/store/pgmigrations/002_audit_trail.sql new file mode 100644 index 00000000..77755d6e --- /dev/null +++ b/internal/store/pgmigrations/002_audit_trail.sql @@ -0,0 +1,17 @@ +-- Audit trail: extend activities with IP/UA, nullable workspace, expanded action types. + +-- Add new columns +ALTER TABLE activities ADD COLUMN IF NOT EXISTS ip_address TEXT; +ALTER TABLE activities ADD COLUMN IF NOT EXISTS user_agent TEXT; + +-- Make workspace_id nullable (auth events have no workspace) +ALTER TABLE activities ALTER COLUMN workspace_id DROP NOT NULL; + +-- Drop restrictive CHECK constraints and replace with open ones +ALTER TABLE activities DROP CONSTRAINT IF EXISTS activities_action_check; +ALTER TABLE activities DROP CONSTRAINT IF EXISTS activities_actor_check; +ALTER TABLE activities DROP CONSTRAINT IF EXISTS activities_source_check; + +-- Add indexes for audit log queries +CREATE INDEX IF NOT EXISTS idx_activities_action ON activities(action, created_at); +CREATE INDEX IF NOT EXISTS idx_activities_user ON activities(user_id, created_at); diff --git a/internal/store/store.go b/internal/store/store.go index 52a21c95..ed0b4768 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -138,6 +138,7 @@ func (s *Store) migrate() error { "019_agent_roles_tools.sql", "020_role_sort_order.sql", "021_phase_to_links.sql", + "022_audit_trail.sql", } for _, name := range migrations { @@ -183,6 +184,7 @@ func (s *Store) migratePostgres() error { migrations := []string{ "001_initial.sql", + "002_audit_trail.sql", } for _, name := range migrations {