serve.go

v1.2.0

package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"time"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/spf13/cobra"
	"golang.org/x/mod/sumdb"

	"example.com/curator/internal/admin"
	"example.com/curator/internal/config"
	"example.com/curator/internal/dochtml"
	"example.com/curator/internal/git"
	"example.com/curator/internal/metrics"
	"example.com/curator/internal/oidcauth"
	"example.com/curator/internal/server"
	"example.com/curator/internal/storage"
	"example.com/curator/internal/store"
	isumdb "example.com/curator/internal/sumdb"
)

func serveCmd() *cobra.Command {
	var configPath string

	cmd := &cobra.Command{
		Use:   "serve",
		Short: "Start the curator HTTP server",
		RunE: func(cmd *cobra.Command, args []string) error {
			return runServe(configPath)
		},
	}

	cmd.Flags().StringVarP(&configPath, "config", "c", "", "path to config file (optional, env vars always apply)")

	return cmd
}

func runServe(configPath string) error {
	cfg, err := config.Load(configPath)
	if err != nil {
		return err
	}

	log.Printf("configuration:\n%s", cfg.Dump())

	// Open module store.
	moduleStore, err := store.Open(cfg.Database.Type, cfg.Database.DSN)
	if err != nil {
		return fmt.Errorf("open store: %w", err)
	}
	defer moduleStore.Close()

	docRenderer, err := dochtml.NewRenderer()
	if err != nil {
		return fmt.Errorf("init doc renderer: %w", err)
	}

	srv := &server.Server{
		Cfg:         cfg,
		Resolver:    moduleStore,
		Credentials: moduleStore,
		Git:         git.NewCache(cfg.Cache),
		HTTPClient:  &http.Client{Timeout: 30 * time.Second},
		DocRenderer: docRenderer,
	}

	// Load auth tokens if configured.
	if cfg.AuthTokens != "" {
		tokens, err := server.ParseAuthTokens(cfg.AuthTokens)
		if err != nil {
			return fmt.Errorf("load auth tokens: %w", err)
		}

		srv.AuthTokens = tokens
		log.Printf("loaded %d auth tokens", len(tokens))
	}

	// Initialize storage if configured.
	if cfg.Storage != nil {
		st, err := initStorage(context.Background(), cfg.Storage)
		if err != nil {
			return fmt.Errorf("init storage: %w", err)
		}

		srv.Store = st
		storageType := cfg.Storage.Type
		if storageType == "" {
			storageType = "s3"
		}
		log.Printf("storage enabled: type=%s", storageType)
	}

	// Initialize sumdb if configured.
	var sumdbOps *isumdb.Ops

	if cfg.Sumdb != nil && cfg.Sumdb.Enabled {
		signer, err := config.ResolveValue(cfg.Sumdb.Key)
		if err != nil {
			return fmt.Errorf("read sumdb key: %w", err)
		}
		sumdbOps = isumdb.NewOps(cfg.Host, moduleStore, moduleStore, srv.Git, moduleStore, signer)
		log.Printf("sumdb enabled with %d records", sumdbOps.RecordCount())
	}

	// Metrics.
	metrics.Register()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var bucketStats metrics.BucketStatsFunc
	if srv.Store != nil {
		if sp, ok := srv.Store.(storage.StatsProvider); ok {
			bucketStats = sp.BucketStats
		}
	}

	var recordCount metrics.RecordCountFunc
	if sumdbOps != nil {
		recordCount = sumdbOps.RecordCount
	}

	moduleCount := func() int { return moduleStore.ModuleCount() }
	metrics.StartCollector(ctx, cfg.Cache, moduleCount, bucketStats, recordCount, 30*time.Second)

	// Instrumentation server (metrics + health) on a separate port.
	if inst := cfg.Instrumentation; inst != nil && inst.Listen != "" {
		instrMux := http.NewServeMux()

		metricsPath := inst.MetricsPath
		if metricsPath == "" {
			metricsPath = "/metrics"
		}
		instrMux.Handle(metricsPath, promhttp.Handler())

		healthPath := inst.HealthPath
		if healthPath == "" {
			healthPath = "/healthz"
		}
		instrMux.HandleFunc(healthPath, healthHandler)

		go func() {
			log.Printf("instrumentation serving on %s (metrics: %s, health: %s)",
				inst.Listen, metricsPath, healthPath)
			if err := http.ListenAndServe(inst.Listen, instrMux); err != nil {
				log.Fatalf("instrumentation server: %v", err)
			}
		}()
	}

	// Main HTTP mux.
	mux := http.NewServeMux()
	mux.Handle("/-/static/", http.StripPrefix("/-/static/", http.FileServer(dochtml.StaticFS())))

	// Admin UI and API.
	adminHandler, err := admin.New(moduleStore, version)
	if err != nil {
		return fmt.Errorf("init admin: %w", err)
	}

	apiHandler := admin.NewAPI(moduleStore)

	adminToken := cfg.AdminToken
	if adminToken != "" {
		parsed, err := server.ParseAdminToken(adminToken)
		if err != nil {
			return fmt.Errorf("parse admin token: %w", err)
		}
		adminToken = parsed
	}

	// Initialize OIDC if configured.
	var oidcProvider *oidcauth.Provider
	if cfg.OIDC != nil && cfg.OIDC.Issuer != "" {
		secret, err := config.ResolveValue(cfg.OIDC.ClientSecret)
		if err != nil {
			return fmt.Errorf("resolve oidc client_secret: %w", err)
		}
		oidcProvider, err = oidcauth.New(context.Background(), cfg.OIDC, secret)
		if err != nil {
			return fmt.Errorf("init oidc: %w", err)
		}
		mux.HandleFunc("/-/oidc/login", oidcProvider.LoginHandler)
		mux.HandleFunc("/-/oidc/callback", oidcProvider.CallbackHandler)
		mux.HandleFunc("/-/oidc/logout", oidcProvider.LogoutHandler)
		log.Printf("OIDC enabled: issuer=%s client_id=%s", cfg.OIDC.Issuer, cfg.OIDC.ClientID)
	}

	mux.Handle("/-/admin/", admin.AuthMiddleware(adminToken, oidcProvider, adminHandler))
	mux.Handle("/-/api/", admin.AuthMiddleware(adminToken, oidcProvider, apiHandler))

	if sumdbOps != nil {
		sumdbSrv := sumdb.NewServer(sumdbOps)
		prefix := "/sumdb/" + cfg.Host

		var upstream string
		if cfg.Sumdb.Upstream != "" {
			upstream = cfg.Sumdb.Upstream
			log.Printf("sumdb upstream proxy: %s", upstream)
		}

		handler := srv.SumdbHandler(
			http.StripPrefix(prefix, sumdbSrv),
			upstream, prefix,
		)
		mux.Handle(prefix+"/", srv.SumdbAuthMiddleware(handler))
		log.Printf("sumdb serving at %s/", prefix)
	}

	mux.Handle("/", srv)

	log.Printf("curator serving on %s (host: %s, db: %s/%s, modules: %d)",
		cfg.Listen, cfg.Host, cfg.Database.Type, cfg.Database.DSN, moduleStore.ModuleCount())

	return http.ListenAndServe(cfg.Listen, mux)
}

func initStorage(ctx context.Context, cfg *config.StorageConfig) (storage.Storage, error) {
	switch cfg.Type {
	case "s3", "":
		return storage.NewS3Storage(ctx, cfg.S3())
	case "disk":
		return storage.NewDiskStorage(cfg.Root)
	case "memory":
		return storage.NewMemoryStorage(cfg.MaxBytes), nil
	default:
		return nil, fmt.Errorf("unknown storage type: %q", cfg.Type)
	}
}

func healthHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	w.Write([]byte(`{"status":"ok"}` + "\n"))
}


1	package main
2
3	import (
4	"context"
5	"fmt"
6	"log"
7	"net/http"
8	"time"
9
10	"github.com/prometheus/client_golang/prometheus/promhttp"
11	"github.com/spf13/cobra"
12	"golang.org/x/mod/sumdb"
13
14	"example.com/curator/internal/admin"
15	"example.com/curator/internal/config"
16	"example.com/curator/internal/dochtml"
17	"example.com/curator/internal/git"
18	"example.com/curator/internal/metrics"
19	"example.com/curator/internal/oidcauth"
20	"example.com/curator/internal/server"
21	"example.com/curator/internal/storage"
22	"example.com/curator/internal/store"
23	isumdb "example.com/curator/internal/sumdb"
24	)
25
26	func serveCmd() *cobra.Command {
27	var configPath string
28
29	cmd := &cobra.Command{
30	Use: "serve",
31	Short: "Start the curator HTTP server",
32	RunE: func(cmd *cobra.Command, args []string) error {
33	return runServe(configPath)
34	},
35	}
36
37	cmd.Flags().StringVarP(&configPath, "config", "c", "", "path to config file (optional, env vars always apply)")
38
39	return cmd
40	}
41
42	func runServe(configPath string) error {
43	cfg, err := config.Load(configPath)
44	if err != nil {
45	return err
46	}
47
48	log.Printf("configuration:\n%s", cfg.Dump())
49
50	// Open module store.
51	moduleStore, err := store.Open(cfg.Database.Type, cfg.Database.DSN)
52	if err != nil {
53	return fmt.Errorf("open store: %w", err)
54	}
55	defer moduleStore.Close()
56
57	docRenderer, err := dochtml.NewRenderer()
58	if err != nil {
59	return fmt.Errorf("init doc renderer: %w", err)
60	}
61
62	srv := &server.Server{
63	Cfg: cfg,
64	Resolver: moduleStore,
65	Credentials: moduleStore,
66	Git: git.NewCache(cfg.Cache),
67	HTTPClient: &http.Client{Timeout: 30 * time.Second},
68	DocRenderer: docRenderer,
69	}
70
71	// Load auth tokens if configured.
72	if cfg.AuthTokens != "" {
73	tokens, err := server.ParseAuthTokens(cfg.AuthTokens)
74	if err != nil {
75	return fmt.Errorf("load auth tokens: %w", err)
76	}
77
78	srv.AuthTokens = tokens
79	log.Printf("loaded %d auth tokens", len(tokens))
80	}
81
82	// Initialize storage if configured.
83	if cfg.Storage != nil {
84	st, err := initStorage(context.Background(), cfg.Storage)
85	if err != nil {
86	return fmt.Errorf("init storage: %w", err)
87	}
88
89	srv.Store = st
90	storageType := cfg.Storage.Type
91	if storageType == "" {
92	storageType = "s3"
93	}
94	log.Printf("storage enabled: type=%s", storageType)
95	}
96
97	// Initialize sumdb if configured.
98	var sumdbOps *isumdb.Ops
99
100	if cfg.Sumdb != nil && cfg.Sumdb.Enabled {
101	signer, err := config.ResolveValue(cfg.Sumdb.Key)
102	if err != nil {
103	return fmt.Errorf("read sumdb key: %w", err)
104	}
105	sumdbOps = isumdb.NewOps(cfg.Host, moduleStore, moduleStore, srv.Git, moduleStore, signer)
106	log.Printf("sumdb enabled with %d records", sumdbOps.RecordCount())
107	}
108
109	// Metrics.
110	metrics.Register()
111
112	ctx, cancel := context.WithCancel(context.Background())
113	defer cancel()
114
115	var bucketStats metrics.BucketStatsFunc
116	if srv.Store != nil {
117	if sp, ok := srv.Store.(storage.StatsProvider); ok {
118	bucketStats = sp.BucketStats
119	}
120	}
121
122	var recordCount metrics.RecordCountFunc
123	if sumdbOps != nil {
124	recordCount = sumdbOps.RecordCount
125	}
126
127	moduleCount := func() int { return moduleStore.ModuleCount() }
128	metrics.StartCollector(ctx, cfg.Cache, moduleCount, bucketStats, recordCount, 30*time.Second)
129
130	// Instrumentation server (metrics + health) on a separate port.
131	if inst := cfg.Instrumentation; inst != nil && inst.Listen != "" {
132	instrMux := http.NewServeMux()
133
134	metricsPath := inst.MetricsPath
135	if metricsPath == "" {
136	metricsPath = "/metrics"
137	}
138	instrMux.Handle(metricsPath, promhttp.Handler())
139
140	healthPath := inst.HealthPath
141	if healthPath == "" {
142	healthPath = "/healthz"
143	}
144	instrMux.HandleFunc(healthPath, healthHandler)
145
146	go func() {
147	log.Printf("instrumentation serving on %s (metrics: %s, health: %s)",
148	inst.Listen, metricsPath, healthPath)
149	if err := http.ListenAndServe(inst.Listen, instrMux); err != nil {
150	log.Fatalf("instrumentation server: %v", err)
151	}
152	}()
153	}
154
155	// Main HTTP mux.
156	mux := http.NewServeMux()
157	mux.Handle("/-/static/", http.StripPrefix("/-/static/", http.FileServer(dochtml.StaticFS())))
158
159	// Admin UI and API.
160	adminHandler, err := admin.New(moduleStore, version)
161	if err != nil {
162	return fmt.Errorf("init admin: %w", err)
163	}
164
165	apiHandler := admin.NewAPI(moduleStore)
166
167	adminToken := cfg.AdminToken
168	if adminToken != "" {
169	parsed, err := server.ParseAdminToken(adminToken)
170	if err != nil {
171	return fmt.Errorf("parse admin token: %w", err)
172	}
173	adminToken = parsed
174	}
175
176	// Initialize OIDC if configured.
177	var oidcProvider *oidcauth.Provider
178	if cfg.OIDC != nil && cfg.OIDC.Issuer != "" {
179	secret, err := config.ResolveValue(cfg.OIDC.ClientSecret)
180	if err != nil {
181	return fmt.Errorf("resolve oidc client_secret: %w", err)
182	}
183	oidcProvider, err = oidcauth.New(context.Background(), cfg.OIDC, secret)
184	if err != nil {
185	return fmt.Errorf("init oidc: %w", err)
186	}
187	mux.HandleFunc("/-/oidc/login", oidcProvider.LoginHandler)
188	mux.HandleFunc("/-/oidc/callback", oidcProvider.CallbackHandler)
189	mux.HandleFunc("/-/oidc/logout", oidcProvider.LogoutHandler)
190	log.Printf("OIDC enabled: issuer=%s client_id=%s", cfg.OIDC.Issuer, cfg.OIDC.ClientID)
191	}
192
193	mux.Handle("/-/admin/", admin.AuthMiddleware(adminToken, oidcProvider, adminHandler))
194	mux.Handle("/-/api/", admin.AuthMiddleware(adminToken, oidcProvider, apiHandler))
195
196	if sumdbOps != nil {
197	sumdbSrv := sumdb.NewServer(sumdbOps)
198	prefix := "/sumdb/" + cfg.Host
199
200	var upstream string
201	if cfg.Sumdb.Upstream != "" {
202	upstream = cfg.Sumdb.Upstream
203	log.Printf("sumdb upstream proxy: %s", upstream)
204	}
205
206	handler := srv.SumdbHandler(
207	http.StripPrefix(prefix, sumdbSrv),
208	upstream, prefix,
209	)
210	mux.Handle(prefix+"/", srv.SumdbAuthMiddleware(handler))
211	log.Printf("sumdb serving at %s/", prefix)
212	}
213
214	mux.Handle("/", srv)
215
216	log.Printf("curator serving on %s (host: %s, db: %s/%s, modules: %d)",
217	cfg.Listen, cfg.Host, cfg.Database.Type, cfg.Database.DSN, moduleStore.ModuleCount())
218
219	return http.ListenAndServe(cfg.Listen, mux)
220	}
221
222	func initStorage(ctx context.Context, cfg *config.StorageConfig) (storage.Storage, error) {
223	switch cfg.Type {
224	case "s3", "":
225	return storage.NewS3Storage(ctx, cfg.S3())
226	case "disk":
227	return storage.NewDiskStorage(cfg.Root)
228	case "memory":
229	return storage.NewMemoryStorage(cfg.MaxBytes), nil
230	default:
231	return nil, fmt.Errorf("unknown storage type: %q", cfg.Type)
232	}
233	}
234
235	func healthHandler(w http.ResponseWriter, r *http.Request) {
236	w.Header().Set("Content-Type", "application/json")
237	w.Write([]byte(`{"status":"ok"}` + "\n"))
238	}
239

serve.go

Source Files