|
| 1 | +package main |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "fmt" |
| 6 | + "log/slog" |
| 7 | + "net/http" |
| 8 | + "os" |
| 9 | + "runtime" |
| 10 | + |
| 11 | + "github.com/did-method-plc/go-didplc/replica" |
| 12 | + "github.com/prometheus/client_golang/prometheus/promhttp" |
| 13 | + "github.com/urfave/cli/v3" |
| 14 | + "golang.org/x/sync/errgroup" |
| 15 | +) |
| 16 | + |
| 17 | +func main() { |
| 18 | + cmd := &cli.Command{ |
| 19 | + Name: "plc-replica", |
| 20 | + Usage: "PLC directory replica server", |
| 21 | + Flags: []cli.Flag{ |
| 22 | + &cli.StringFlag{ |
| 23 | + Name: "db-url", |
| 24 | + Usage: "Database URL (e.g. sqlite://replica.db?_journal_mode=WAL, postgres://user:pass@host/db)", |
| 25 | + Value: "sqlite://replica.db?mode=rwc&cache=shared&_journal_mode=WAL", |
| 26 | + Sources: cli.EnvVars("DATABASE_URL"), |
| 27 | + }, |
| 28 | + &cli.StringFlag{ |
| 29 | + Name: "bind", |
| 30 | + Usage: "HTTP server listen address", |
| 31 | + Value: ":6780", |
| 32 | + Sources: cli.EnvVars("REPLICA_BIND"), |
| 33 | + }, |
| 34 | + &cli.StringFlag{ |
| 35 | + Name: "metrics-addr", |
| 36 | + Usage: "Metrics HTTP server listen address", |
| 37 | + Value: ":9464", |
| 38 | + Sources: cli.EnvVars("METRICS_ADDR"), |
| 39 | + }, |
| 40 | + &cli.BoolFlag{ |
| 41 | + Name: "no-ingest", |
| 42 | + Usage: "Disable ingestion from upstream directory", |
| 43 | + Sources: cli.EnvVars("NO_INGEST"), |
| 44 | + }, |
| 45 | + &cli.StringFlag{ |
| 46 | + Name: "upstream-directory-url", |
| 47 | + Usage: "Upstream PLC directory base URL", |
| 48 | + Value: "https://plc.directory", |
| 49 | + Sources: cli.EnvVars("UPSTREAM_DIRECTORY_URL"), |
| 50 | + }, |
| 51 | + &cli.Int64Flag{ |
| 52 | + Name: "cursor-override", |
| 53 | + Usage: "Initial cursor value used to sync from the upstream host. May be useful when switching the upstream host", |
| 54 | + Value: -1, |
| 55 | + Sources: cli.EnvVars("CURSOR_OVERRIDE"), |
| 56 | + }, |
| 57 | + &cli.IntFlag{ |
| 58 | + Name: "num-workers", |
| 59 | + Usage: "Number of validation worker threads (0 = auto)", |
| 60 | + Value: 0, |
| 61 | + Sources: cli.EnvVars("NUM_WORKERS"), |
| 62 | + }, |
| 63 | + &cli.StringFlag{ |
| 64 | + Name: "log-level", |
| 65 | + Usage: "Log level (debug, info, warn, error)", |
| 66 | + Value: "info", |
| 67 | + Sources: cli.EnvVars("LOG_LEVEL"), |
| 68 | + }, |
| 69 | + &cli.BoolFlag{ |
| 70 | + Name: "log-json", |
| 71 | + Usage: "Output logs in JSON format", |
| 72 | + Sources: cli.EnvVars("LOG_JSON"), |
| 73 | + }, |
| 74 | + }, |
| 75 | + Action: run, |
| 76 | + } |
| 77 | + |
| 78 | + if err := cmd.Run(context.Background(), os.Args); err != nil { |
| 79 | + slog.Error("fatal error", "error", err) |
| 80 | + os.Exit(1) |
| 81 | + } |
| 82 | +} |
| 83 | + |
| 84 | +func run(ctx context.Context, cmd *cli.Command) error { |
| 85 | + // Parse configuration |
| 86 | + dbURL := cmd.String("db-url") |
| 87 | + httpAddr := cmd.String("bind") |
| 88 | + metricsAddr := cmd.String("metrics-addr") |
| 89 | + noIngest := cmd.Bool("no-ingest") |
| 90 | + directoryURL := cmd.String("upstream-directory-url") |
| 91 | + cursorOverride := cmd.Int64("cursor-override") |
| 92 | + numWorkers := cmd.Int("num-workers") |
| 93 | + logLevel := cmd.String("log-level") |
| 94 | + logJSON := cmd.Bool("log-json") |
| 95 | + |
| 96 | + // Initialize logger |
| 97 | + var level slog.Level |
| 98 | + switch logLevel { |
| 99 | + case "debug": |
| 100 | + level = slog.LevelDebug |
| 101 | + case "info": |
| 102 | + level = slog.LevelInfo |
| 103 | + case "warn": |
| 104 | + level = slog.LevelWarn |
| 105 | + case "error": |
| 106 | + level = slog.LevelError |
| 107 | + default: |
| 108 | + level = slog.LevelInfo |
| 109 | + } |
| 110 | + |
| 111 | + var handler slog.Handler |
| 112 | + opts := &slog.HandlerOptions{Level: level} |
| 113 | + if logJSON { |
| 114 | + handler = slog.NewJSONHandler(os.Stdout, opts) |
| 115 | + } else { |
| 116 | + handler = slog.NewTextHandler(os.Stdout, opts) |
| 117 | + } |
| 118 | + logger := slog.New(handler) |
| 119 | + slog.SetDefault(logger) |
| 120 | + |
| 121 | + if numWorkers <= 0 { |
| 122 | + numWorkers = runtime.NumCPU() |
| 123 | + } |
| 124 | + |
| 125 | + otelShutdown, err := setupOTel(ctx) |
| 126 | + if err != nil { |
| 127 | + return fmt.Errorf("otel setup: %w", err) |
| 128 | + } |
| 129 | + defer otelShutdown(context.Background()) |
| 130 | + |
| 131 | + store, err := replica.NewGormOpStore(dbURL, logger) |
| 132 | + if err != nil { |
| 133 | + return fmt.Errorf("failed to create store: %w", err) |
| 134 | + } |
| 135 | + |
| 136 | + state := replica.NewReplicaState() |
| 137 | + server := replica.NewServer(store, state, httpAddr, logger) |
| 138 | + g, gctx := errgroup.WithContext(ctx) |
| 139 | + |
| 140 | + g.Go(server.Run) |
| 141 | + |
| 142 | + g.Go(func() error { |
| 143 | + mux := http.NewServeMux() |
| 144 | + mux.Handle("/metrics", promhttp.Handler()) |
| 145 | + slog.Info("metrics server listening", "addr", metricsAddr) |
| 146 | + return http.ListenAndServe(metricsAddr, mux) |
| 147 | + }) |
| 148 | + |
| 149 | + if !noIngest { |
| 150 | + ingestor, err := replica.NewIngestor(store, state, directoryURL, cursorOverride, numWorkers, logger) |
| 151 | + if err != nil { |
| 152 | + return err |
| 153 | + } |
| 154 | + g.Go(func() error { |
| 155 | + return ingestor.Run(gctx) |
| 156 | + }) |
| 157 | + } |
| 158 | + |
| 159 | + return g.Wait() |
| 160 | +} |
0 commit comments