Bluemage/go/cmd/bluemage/serve/serve.go

163 lines
4.5 KiB
Go
Raw Normal View History

package serve
import (
"context"
"errors"
2024-08-11 19:50:43 +07:00
"fmt"
"log/slog"
2024-08-11 19:50:43 +07:00
"net"
"net/http"
2024-08-07 10:41:00 +07:00
"os"
"path/filepath"
"time"
"connectrpc.com/connect"
2024-08-12 20:20:12 +07:00
"connectrpc.com/otelconnect"
2024-08-08 20:49:18 +07:00
"connectrpc.com/validate"
"github.com/XSAM/otelsql"
2024-08-08 13:41:18 +07:00
sqldblogger "github.com/simukti/sqldb-logger"
"github.com/spf13/cobra"
"github.com/stephenafamo/bob"
"github.com/tigorlazuardi/bluemage/go/api"
2024-08-11 19:50:43 +07:00
"github.com/tigorlazuardi/bluemage/go/config"
v1DeviceConnect "github.com/tigorlazuardi/bluemage/go/gen/proto/device/v1/v1connect"
v1SubredditsConnect "github.com/tigorlazuardi/bluemage/go/gen/proto/subreddits/v1/v1connect"
"github.com/tigorlazuardi/bluemage/go/gen/reddit"
"github.com/tigorlazuardi/bluemage/go/pkg/errs"
2024-08-07 10:41:00 +07:00
"github.com/tigorlazuardi/bluemage/go/pkg/log"
2024-08-12 20:20:12 +07:00
"github.com/tigorlazuardi/bluemage/go/pkg/telemetry"
"github.com/tigorlazuardi/bluemage/go/server"
semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
"golang.org/x/net/http2"
"golang.org/x/net/http2/h2c"
)
var Cmd = &cobra.Command{
Use: "serve",
RunE: func(cmd *cobra.Command, args []string) error {
2024-08-11 19:50:43 +07:00
ctx := cmd.Context()
cfg := config.FromContext(ctx)
var cleanups []func() error
defer func() {
var e []error
for _, c := range cleanups {
e = append(e, c())
}
if err := errors.Join(e...); err != nil {
fmt.Println(err.Error())
}
}()
logHandler, cleanup := log.NewHandler(cfg)
cleanups = append(cleanups, cleanup)
slog.SetDefault(slog.New(logHandler))
2024-08-11 19:50:43 +07:00
2024-08-12 20:20:12 +07:00
tele, err := telemetry.New(ctx, cfg)
if err != nil {
return errs.Wrap(err, "failed to create telemetry")
}
cleanups = append(cleanups, tele.Close)
2024-08-11 19:50:43 +07:00
if err := os.MkdirAll(cfg.String("download.directory"), 0755); err != nil {
return errs.Wrapw(err, "failed to create download directory", "path", cfg.String("download.directory"))
}
dbpath := cfg.String("db.path")
dir := filepath.Dir(dbpath)
if err := os.MkdirAll(dir, 0755); err != nil {
return errs.Wrapw(err, "failed to create database directory", "path", dir)
}
dsn := fmt.Sprintf("file:%s", cfg.String("db.path"))
sqldb, err := otelsql.Open(cfg.String("db.driver"), dsn, otelsql.WithAttributes(semconv.DBSystemSqlite))
if err != nil {
2024-08-11 19:50:43 +07:00
return errs.Wrapw(err, "failed to open database",
"driver", cfg.String("db.driver"),
"dsn", dsn,
2024-08-11 19:50:43 +07:00
)
}
cleanups = append(cleanups, sqldb.Close)
if err := otelsql.RegisterDBStatsMetrics(sqldb, otelsql.WithAttributes(semconv.DBSystemSqlite)); err != nil {
return errs.Wrapw(
err, "failed to register database stats metrics",
"driver", cfg.String("db.driver"),
"dsn", dsn,
)
}
2024-08-08 13:41:18 +07:00
sqldb = sqldblogger.OpenDriver(
dsn,
2024-08-08 13:41:18 +07:00
sqldb.Driver(),
log.SQLLogger{},
sqldblogger.WithSQLQueryAsMessage(true),
)
db := bob.New(sqldb)
2024-08-07 10:41:00 +07:00
client, err := reddit.NewClient("https://reddit.com", reddit.WithClient(&http.Client{
Transport: log.NewRoundTripper(http.DefaultTransport),
}))
if err != nil {
panic(err)
}
api := &api.API{
Executor: db,
DB: sqldb,
Reddit: client,
}
2024-08-08 20:49:18 +07:00
validationInterceptor, err := validate.NewInterceptor()
if err != nil {
return errs.Wrap(err, "failed to create validation interceptor")
}
2024-08-12 20:20:12 +07:00
otelInterceptor, err := otelconnect.NewInterceptor()
if err != nil {
return errs.Wrap(err, "failed to create otel interceptor")
}
interceptors := []connect.Interceptor{
2024-08-08 20:49:18 +07:00
validationInterceptor,
2024-08-12 20:20:12 +07:00
otelInterceptor,
server.ErrorMessageInterceptor(),
2024-08-08 20:49:18 +07:00
server.LogInterceptor(),
}
handlerOpts := []connect.HandlerOption{
connect.WithInterceptors(interceptors...),
}
mux := http.NewServeMux()
mux.Handle(v1DeviceConnect.NewDeviceServiceHandler(&server.DeviceHandler{API: api}, handlerOpts...))
mux.Handle(v1SubredditsConnect.NewSubredditsServiceHandler(&server.SubredditHandler{API: api}, handlerOpts...))
server := &http.Server{
2024-08-11 19:50:43 +07:00
Addr: fmt.Sprintf("%s:%s", cfg.String("http.host"), cfg.String("http.port")),
Handler: h2c.NewHandler(server.WithCORS(mux), &http2.Server{}),
2024-08-11 19:50:43 +07:00
BaseContext: func(net.Listener) context.Context {
return ctx
},
}
go func() {
2024-08-11 19:50:43 +07:00
<-ctx.Done()
slog.Info("Exit signal received. Shutting down server")
shutdownCtx, cancel := context.WithTimeout(context.Background(), time.Second*5)
defer cancel()
_ = server.Shutdown(shutdownCtx)
}()
slog.Info("ConnectRPC server started", "addr", server.Addr)
err = server.ListenAndServe()
if err != nil && !errors.Is(err, http.ErrServerClosed) {
return errs.Wrap(err, "failed to serve")
}
2024-08-08 22:51:53 +07:00
slog.Info("ConnectRPC server stopped")
return nil
},
SilenceUsage: true,
}
2024-08-11 19:50:43 +07:00
func Register(cmd *cobra.Command, config *config.Config) {
}