serve.go

v1.1.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,
		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, 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	Git: git.NewCache(cfg.Cache),
66	HTTPClient: &http.Client{Timeout: 30 * time.Second},
67	DocRenderer: docRenderer,
68	}
69
70	// Load auth tokens if configured.
71	if cfg.AuthTokens != "" {
72	tokens, err := server.ParseAuthTokens(cfg.AuthTokens)
73	if err != nil {
74	return fmt.Errorf("load auth tokens: %w", err)
75	}
76
77	srv.AuthTokens = tokens
78	log.Printf("loaded %d auth tokens", len(tokens))
79	}
80
81	// Initialize storage if configured.
82	if cfg.Storage != nil {
83	st, err := initStorage(context.Background(), cfg.Storage)
84	if err != nil {
85	return fmt.Errorf("init storage: %w", err)
86	}
87
88	srv.Store = st
89	storageType := cfg.Storage.Type
90	if storageType == "" {
91	storageType = "s3"
92	}
93	log.Printf("storage enabled: type=%s", storageType)
94	}
95
96	// Initialize sumdb if configured.
97	var sumdbOps *isumdb.Ops
98
99	if cfg.Sumdb != nil && cfg.Sumdb.Enabled {
100	signer, err := config.ResolveValue(cfg.Sumdb.Key)
101	if err != nil {
102	return fmt.Errorf("read sumdb key: %w", err)
103	}
104	sumdbOps = isumdb.NewOps(cfg.Host, moduleStore, srv.Git, moduleStore, signer)
105	log.Printf("sumdb enabled with %d records", sumdbOps.RecordCount())
106	}
107
108	// Metrics.
109	metrics.Register()
110
111	ctx, cancel := context.WithCancel(context.Background())
112	defer cancel()
113
114	var bucketStats metrics.BucketStatsFunc
115	if srv.Store != nil {
116	if sp, ok := srv.Store.(storage.StatsProvider); ok {
117	bucketStats = sp.BucketStats
118	}
119	}
120
121	var recordCount metrics.RecordCountFunc
122	if sumdbOps != nil {
123	recordCount = sumdbOps.RecordCount
124	}
125
126	moduleCount := func() int { return moduleStore.ModuleCount() }
127	metrics.StartCollector(ctx, cfg.Cache, moduleCount, bucketStats, recordCount, 30*time.Second)
128
129	// Instrumentation server (metrics + health) on a separate port.
130	if inst := cfg.Instrumentation; inst != nil && inst.Listen != "" {
131	instrMux := http.NewServeMux()
132
133	metricsPath := inst.MetricsPath
134	if metricsPath == "" {
135	metricsPath = "/metrics"
136	}
137	instrMux.Handle(metricsPath, promhttp.Handler())
138
139	healthPath := inst.HealthPath
140	if healthPath == "" {
141	healthPath = "/healthz"
142	}
143	instrMux.HandleFunc(healthPath, healthHandler)
144
145	go func() {
146	log.Printf("instrumentation serving on %s (metrics: %s, health: %s)",
147	inst.Listen, metricsPath, healthPath)
148	if err := http.ListenAndServe(inst.Listen, instrMux); err != nil {
149	log.Fatalf("instrumentation server: %v", err)
150	}
151	}()
152	}
153
154	// Main HTTP mux.
155	mux := http.NewServeMux()
156	mux.Handle("/-/static/", http.StripPrefix("/-/static/", http.FileServer(dochtml.StaticFS())))
157
158	// Admin UI and API.
159	adminHandler, err := admin.New(moduleStore, version)
160	if err != nil {
161	return fmt.Errorf("init admin: %w", err)
162	}
163
164	apiHandler := admin.NewAPI(moduleStore)
165
166	adminToken := cfg.AdminToken
167	if adminToken != "" {
168	parsed, err := server.ParseAdminToken(adminToken)
169	if err != nil {
170	return fmt.Errorf("parse admin token: %w", err)
171	}
172	adminToken = parsed
173	}
174
175	// Initialize OIDC if configured.
176	var oidcProvider *oidcauth.Provider
177	if cfg.OIDC != nil && cfg.OIDC.Issuer != "" {
178	secret, err := config.ResolveValue(cfg.OIDC.ClientSecret)
179	if err != nil {
180	return fmt.Errorf("resolve oidc client_secret: %w", err)
181	}
182	oidcProvider, err = oidcauth.New(context.Background(), cfg.OIDC, secret)
183	if err != nil {
184	return fmt.Errorf("init oidc: %w", err)
185	}
186	mux.HandleFunc("/-/oidc/login", oidcProvider.LoginHandler)
187	mux.HandleFunc("/-/oidc/callback", oidcProvider.CallbackHandler)
188	mux.HandleFunc("/-/oidc/logout", oidcProvider.LogoutHandler)
189	log.Printf("OIDC enabled: issuer=%s client_id=%s", cfg.OIDC.Issuer, cfg.OIDC.ClientID)
190	}
191
192	mux.Handle("/-/admin/", admin.AuthMiddleware(adminToken, oidcProvider, adminHandler))
193	mux.Handle("/-/api/", admin.AuthMiddleware(adminToken, oidcProvider, apiHandler))
194
195	if sumdbOps != nil {
196	sumdbSrv := sumdb.NewServer(sumdbOps)
197	prefix := "/sumdb/" + cfg.Host
198
199	var upstream string
200	if cfg.Sumdb.Upstream != "" {
201	upstream = cfg.Sumdb.Upstream
202	log.Printf("sumdb upstream proxy: %s", upstream)
203	}
204
205	handler := srv.SumdbHandler(
206	http.StripPrefix(prefix, sumdbSrv),
207	upstream, prefix,
208	)
209	mux.Handle(prefix+"/", srv.SumdbAuthMiddleware(handler))
210	log.Printf("sumdb serving at %s/", prefix)
211	}
212
213	mux.Handle("/", srv)
214
215	log.Printf("curator serving on %s (host: %s, db: %s/%s, modules: %d)",
216	cfg.Listen, cfg.Host, cfg.Database.Type, cfg.Database.DSN, moduleStore.ModuleCount())
217
218	return http.ListenAndServe(cfg.Listen, mux)
219	}
220
221	func initStorage(ctx context.Context, cfg *config.StorageConfig) (storage.Storage, error) {
222	switch cfg.Type {
223	case "s3", "":
224	return storage.NewS3Storage(ctx, cfg.S3())
225	case "disk":
226	return storage.NewDiskStorage(cfg.Root)
227	case "memory":
228	return storage.NewMemoryStorage(cfg.MaxBytes), nil
229	default:
230	return nil, fmt.Errorf("unknown storage type: %q", cfg.Type)
231	}
232	}
233
234	func healthHandler(w http.ResponseWriter, r *http.Request) {
235	w.Header().Set("Content-Type", "application/json")
236	w.Write([]byte(`{"status":"ok"}` + "\n"))
237	}
238

serve.go

Source Files