ws: add mysql-pitr-agent daemon serving binlogs to the server

Agent now runs as a persistent daemon (serve) on the MySQL host: it reads
local binlog files, parses them via mysqlbinlog honouring targeted parse
parameters (binlog files, table, time range, positions), and executes
rollbacks locally. The server no longer accesses binlog files or MySQL
credentials — PITR operations are driven over the existing reverse-WS
tunnel (hub, dispatcher, internal CA) which is now wired into both
binaries: mTLS :9443 listener with file-backed CA, lifecycle hooks for
agent online/offline status, pitr_progress pushes, and pitr_cancel.

Web wizard drops mysql_dsn and selects connected agents; adds optional
targeted parsing fields. docker-compose gains a one-shot provision
service (CA extraction, agent registration, cert issuance, encrypted
config) and scripts/e2e-test.sh drives a full operation through the
agent. deploy/agent.service and install script run serve mode.

Also fixes pre-existing test breakage (chi route contexts, mock
interface drift, cert SAN/key-usage enforcement) across pitr, agent,
audit, ws packages.

Author: Developer

Dockerfile

@@ -43,7 +43,11 @@ RUN npm run build
 # =============================================================================
 FROM alpine:3.20 AS agent
 
-RUN apk add --no-cache ca-certificates tzdata
+# mariadb-client provides mysqlbinlog (symlinked for name compatibility),
+# which the agent uses to parse local binlog files.
+RUN apk add --no-cache ca-certificates tzdata mariadb-client mariadb-server-utils && \
+    ln -sf /usr/bin/mariadb-binlog /usr/bin/mysqlbinlog && \
+    test -x /usr/bin/mysqlbinlog
 
 COPY --from=builder /build/mysql-pitr-agent /usr/local/bin/mysql-pitr-agent
 
@@ -54,14 +58,14 @@ ENTRYPOINT ["mysql-pitr-agent"]
 # =============================================================================
 FROM alpine:3.20 AS server
 
-RUN apk add --no-cache ca-certificates tzdata mariadb-client && \
-    ln -sf /usr/bin/mariadb-binlog /usr/bin/mysqlbinlog && \
-    test -x /usr/bin/mysqlbinlog
+# The server no longer parses binlogs itself — the agent does — so no
+# mysql/mariadb client is needed.
+RUN apk add --no-cache ca-certificates tzdata
 
 COPY --from=builder /build/mysql-pitr-server /usr/local/bin/mysql-pitr-server
 
 COPY --from=frontend /web/dist /usr/share/mysql-pitr/web/
 
-EXPOSE 8080
+EXPOSE 8080 9443
 
 ENTRYPOINT ["mysql-pitr-server"]

cmd/agent/config.go

@@ -0,0 +1,75 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/spf13/cobra"
+
+	"github.com/a-shan/mysql-pitr/internal/config"
+)
+
+// NewConfigCommand creates the `agent config` cobra command tree for
+// encrypting and inspecting the agent config file.
+func NewConfigCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "config",
+		Short: "Encrypt and inspect the agent config file",
+	}
+	cmd.AddCommand(NewConfigEncryptCommand())
+	return cmd
+}
+
+// NewConfigEncryptCommand creates the `agent config encrypt` subcommand that
+// reads a plaintext JSON config and writes the AES-256-GCM encrypted form
+// that `serve` and `flashback --config` load.
+func NewConfigEncryptCommand() *cobra.Command {
+	var (
+		input      string
+		output     string
+		passphrase string
+	)
+
+	cmd := &cobra.Command{
+		Use:   "encrypt",
+		Short: "Encrypt a plaintext JSON config into the on-disk format",
+		Long: `Read a plaintext JSON config (the shape documented in deploy/README.md)
+and write the AES-256-GCM encrypted version that the agent loads with
+--config. The plaintext file is not modified.`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			if input == "" {
+				return fmt.Errorf("--input is required")
+			}
+			if output == "" {
+				return fmt.Errorf("--output is required")
+			}
+			if passphrase == "" {
+				return fmt.Errorf("--passphrase is required")
+			}
+
+			raw, err := os.ReadFile(input)
+			if err != nil {
+				return fmt.Errorf("config encrypt: read input: %w", err)
+			}
+			// Validate the shape early so a bad config never gets encrypted.
+			var cfg config.Config
+			if err := json.Unmarshal(raw, &cfg); err != nil {
+				return fmt.Errorf("config encrypt: input is not a valid agent config: %w", err)
+			}
+			if err := config.SaveConfig(output, passphrase, &cfg); err != nil {
+				return fmt.Errorf("config encrypt: %w", err)
+			}
+			fmt.Printf("wrote encrypted config to %s\n", output)
+			return nil
+		},
+		SilenceUsage: true,
+	}
+
+	flags := cmd.Flags()
+	flags.StringVar(&input, "input", "", "Plaintext JSON config path (required)")
+	flags.StringVar(&output, "output", "", "Encrypted config output path (required)")
+	flags.StringVar(&passphrase, "passphrase", "", "Passphrase for encryption (required)")
+
+	return cmd
+}

cmd/agent/flashback.go

@@ -104,9 +104,14 @@ func RunFlashback(ctx context.Context, opts FlashbackOptions) error {
 	}
 
 	// Resolve binlog directory and build file paths.
-	dataDir, err := resolveDataDir(connCfg)
-	if err != nil {
-		return fmt.Errorf("flashback: resolve binlog directory: %w", err)
+	dataDir, err := conn.GetBinlogDir(ctx)
+	if err != nil || dataDir == "" {
+		// Fall back to a fresh discovery connection (e.g. when the injected
+		// connector cannot resolve the directory).
+		dataDir, err = resolveDataDir(connCfg)
+		if err != nil {
+			return fmt.Errorf("flashback: resolve binlog directory: %w", err)
+		}
 	}
 	log.Printf("MySQL data directory: %s", dataDir)
 
@@ -116,7 +121,10 @@ func RunFlashback(ctx context.Context, opts FlashbackOptions) error {
 	}
 
 	// ---- Parse binlogs via mysqlbinlog ----
-	parseRes, err := parseBinlogWithMySQLBinlog(connCfg, paths, opts.TargetTable, recoveryTime)
+	parseRes, err := mysqlbinlogParse(connCfg, paths, binlogParseOpts{
+		TargetTable: opts.TargetTable,
+		EndTime:     &recoveryTime,
+	})
 	if err != nil {
 		return fmt.Errorf("flashback: parse binlogs: %w", err)
 	}
@@ -232,12 +240,27 @@ func resolveDataDir(cfg connector.ConnConfig) (string, error) {
 	return value, nil
 }
 
+// mysqlbinlogParse is the parse entry point used by RunFlashback. It is a
+// package-level variable so tests can substitute a fake.
+var mysqlbinlogParse = parseBinlogWithMySQLBinlog
+
+// binlogParseOpts controls how a binlog parse is constrained. All fields are
+// optional; a nil/zero field means "no constraint".
+type binlogParseOpts struct {
+	TargetTable     string
+	StartTime       *time.Time
+	EndTime         *time.Time
+	StartPos        uint32
+	StopPos         uint32
+	MySQLBinlogPath string
+}
+
 // parseBinlogWithMySQLBinlog uses the mysqlbinlog tool to parse binlog files
 // and generate reverse SQL statements.
-func parseBinlogWithMySQLBinlog(cfg connector.ConnConfig, paths []string, targetTable string, recoveryTime time.Time) (*connector.ParseResult, error) {
-	parts := strings.SplitN(targetTable, ".", 2)
+func parseBinlogWithMySQLBinlog(cfg connector.ConnConfig, paths []string, opts binlogParseOpts) (*connector.ParseResult, error) {
+	parts := strings.SplitN(opts.TargetTable, ".", 2)
 	if len(parts) != 2 {
-		return nil, fmt.Errorf("invalid target table %q: expected schema.table format", targetTable)
+		return nil, fmt.Errorf("invalid target table %q: expected schema.table format", opts.TargetTable)
 	}
 
 	// Query real column names from MySQL.
@@ -247,16 +270,30 @@ func parseBinlogWithMySQLBinlog(cfg connector.ConnConfig, paths []string, target
 	}
 
 	// Build mysqlbinlog args.
-	recoveryStr := recoveryTime.Format("2006-01-02 15:04:05")
 	args := []string{
 		"--no-defaults",
 		"--base64-output=DECODE-ROWS",
 		"--verbose",
-		"--stop-datetime=" + recoveryStr,
+	}
+	if opts.StartTime != nil {
+		args = append(args, "--start-datetime="+opts.StartTime.Format("2006-01-02 15:04:05"))
+	}
+	if opts.EndTime != nil {
+		args = append(args, "--stop-datetime="+opts.EndTime.Format("2006-01-02 15:04:05"))
+	}
+	if opts.StartPos > 0 {
+		args = append(args, fmt.Sprintf("--start-position=%d", opts.StartPos))
+	}
+	if opts.StopPos > 0 {
+		args = append(args, fmt.Sprintf("--stop-position=%d", opts.StopPos))
 	}
 	args = append(args, paths...)
 
-	cmd := exec.Command("mysqlbinlog", args...)
+	binlogBinary := opts.MySQLBinlogPath
+	if binlogBinary == "" {
+		binlogBinary = "mysqlbinlog"
+	}
+	cmd := exec.Command(binlogBinary, args...)
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {
 		return nil, fmt.Errorf("mysqlbinlog pipe: %w", err)
@@ -336,7 +373,7 @@ func parseBinlogWithMySQLBinlog(cfg connector.ConnConfig, paths []string, target
 	}
 	cmd.Wait()
 
-	log.Printf("mysqlbinlog parsed %d row event(s) for %s", len(events), targetTable)
+	log.Printf("mysqlbinlog parsed %d row event(s) for %s", len(events), opts.TargetTable)
 	return &connector.ParseResult{Events: events, TotalRows: int64(len(events))}, nil
 }
 

cmd/agent/flashback_test.go

@@ -43,6 +43,10 @@ func (m *mockConnector) GetBinlogFiles(ctx context.Context) ([]connector.BinlogF
 	return m.binlogFiles, m.binlogErr
 }
 
+func (m *mockConnector) GetBinlogDir(ctx context.Context) (string, error) {
+	return "/var/lib/mysql/", nil
+}
+
 func (m *mockConnector) ParseBinlog(ctx context.Context, req connector.ParseRequest) (*connector.ParseResult, error) {
 	return m.parseResult, m.parseErr
 }
@@ -60,9 +64,20 @@ func (m *mockConnector) Close() error {
 	return nil
 }
 
+// withFakeParse substitutes the mysqlbinlog parse seam for the duration of a
+// test, returning a fake result so flashback orchestration can be tested
+// without a live MySQL server.
+func withFakeParse(t *testing.T, result *connector.ParseResult, parseErr error) {
+	t.Helper()
+	orig := mysqlbinlogParse
+	mysqlbinlogParse = func(cfg connector.ConnConfig, paths []string, opts binlogParseOpts) (*connector.ParseResult, error) {
+		return result, parseErr
+	}
+	t.Cleanup(func() { mysqlbinlogParse = orig })
+}
+
 // defaultMock creates a mock connector configured for a successful flashback.
-func defaultMock() *mockConnector {
-	return &mockConnector{
+func defaultMock() *mockConnector {	return &mockConnector{
 		preflightResult: &connector.PreflightResult{
 			Status:  connector.PreflightPass,
 			Version: "8.0.32",
@@ -171,6 +186,7 @@ func TestFlashbackCommand_ValidatesRequiredFlags(t *testing.T) {
 
 func TestRunFlashback_DryRun(t *testing.T) {
 	mock := defaultMock()
+	withFakeParse(t, mock.parseResult, nil)
 	opts := FlashbackOptions{
 		Connector:    mock,
 		DSN:          "root:pass@tcp(127.0.0.1:3306)/mydb",
@@ -190,6 +206,7 @@ func TestRunFlashback_OutputFile(t *testing.T) {
 	outputPath := filepath.Join(dir, "rollback.sql")
 
 	mock := defaultMock()
+	withFakeParse(t, mock.parseResult, nil)
 	opts := FlashbackOptions{
 		Connector:    mock,
 		DSN:          "root:pass@tcp(127.0.0.1:3306)/mydb",
@@ -212,6 +229,7 @@ func TestRunFlashback_OutputFile(t *testing.T) {
 
 func TestRunFlashback_Execute(t *testing.T) {
 	mock := defaultMock()
+	withFakeParse(t, mock.parseResult, nil)
 	opts := FlashbackOptions{
 		Connector:    mock,
 		DSN:          "root:pass@tcp(127.0.0.1:3306)/mydb",
@@ -248,10 +266,10 @@ func TestRunFlashback_PreflightFail(t *testing.T) {
 
 func TestRunFlashback_NoEvents(t *testing.T) {
 	mock := defaultMock()
-	mock.parseResult = &connector.ParseResult{
+	withFakeParse(t, &connector.ParseResult{
 		TotalRows: 0,
 		Events:    []connector.RowEvent{},
-	}
+	}, nil)
 
 	opts := FlashbackOptions{
 		Connector:    mock,
@@ -284,6 +302,7 @@ func TestRunFlashback_NoBinlogs(t *testing.T) {
 func TestRunFlashback_ConnectorProvided(t *testing.T) {
 	// When using a pre-injected connector, DSN is not required.
 	mock := defaultMock()
+	withFakeParse(t, mock.parseResult, nil)
 	opts := FlashbackOptions{
 		Connector:    mock,
 		TargetTable:  "mydb.orders",
@@ -425,7 +444,7 @@ func TestNewRootCommand_HasFlashbackSubcommand(t *testing.T) {
 
 func TestRunFlashback_ParseError(t *testing.T) {
 	mock := defaultMock()
-	mock.parseErr = errors.New("binlog file corrupted")
+	withFakeParse(t, nil, errors.New("binlog file corrupted"))
 
 	opts := FlashbackOptions{
 		Connector:    mock,

cmd/agent/main.go

@@ -23,11 +23,15 @@ before a specified recovery time.
 
 Sub-commands:
   flashback    Perform offline binlog flashback (local-only, no WebSocket)
+  serve        Run as a persistent daemon serving the mysql-pitr-server
+  config       Encrypt the agent config file
 `,
 		SilenceUsage: true,
 	}
 
 	cmd.AddCommand(NewFlashbackCommand())
+	cmd.AddCommand(NewServeCommand())
+	cmd.AddCommand(NewConfigCommand())
 
 	return cmd
 }