package deliberi
import (
"context"
"fmt"
"net/http"
"time"
"github.com/go-chi/chi/v5"
"tangled.org/core/deliberi/analytics"
"tangled.org/core/deliberi/config"
deldb "tangled.org/core/deliberi/db"
"tangled.org/core/deliberi/kv"
"tangled.org/core/deliberi/mailer"
delxrpc "tangled.org/core/deliberi/xrpc"
"tangled.org/core/idresolver"
"tangled.org/core/log"
"tangled.org/core/xrpc/serviceauth"
)
func Run(ctx context.Context, cfg *config.Config) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
logger := log.FromContext(ctx)
database, err := deldb.Make(ctx, cfg.DbPath)
if err != nil {
return fmt.Errorf("initializing db: %w", err)
}
resolver := idresolver.DefaultResolver(cfg.PlcUrl)
events := analytics.New(cfg.Posthog, log.SubLogger(logger, "analytics"))
defer events.Close()
var emailKV kv.EmailKV
if cfg.KvApiToken != "" && cfg.CloudflareAccountID != "" && cfg.KvNamespaceID != "" {
emailKV = kv.NewClient(cfg.KvApiToken, cfg.CloudflareAccountID, cfg.KvNamespaceID,
log.SubLogger(logger, "kv"))
}
bobbin := newBobbinClient(cfg.BobbinApiUrl)
ingester := NewIngester(database, bobbin, resolver, cfg.JetstreamEndpoint, log.SubLogger(logger, "ingest"))
ingester.analytics = events
ingester.emailKV = emailKV
ingestDone := make(chan struct{})
go func() {
defer close(ingestDone)
if err := ingester.Run(ctx); err != nil {
logger.Error("ingester stopped", "err", err)
cancel()
}
}()
sender := mailer.New(cfg.Resend, log.SubLogger(logger, "email"))
dispatcher := NewDispatcher(database, sender, cfg.Resend, cfg.BaseURL, resolver, log.SubLogger(logger, "digest"), cfg.Dev)
go dispatcher.Start(ctx)
serviceAuth := serviceauth.NewServiceAuth(logger, resolver.Directory(), serviceauth.DidWeb(cfg.Hostname).String())
x := &delxrpc.Xrpc{
DB: database,
Config: cfg,
Logger: log.SubLogger(logger, "xrpc"),
ServiceAuth: serviceAuth,
IdResolver: resolver,
Sender: sender,
Invites: NewInviteSync(database, bobbin, log.SubLogger(logger, "invites")),
Analytics: events,
KV: emailKV,
}
srv := &http.Server{Addr: cfg.ListenAddr, Handler: chiMount(x)}
go func() {
logger.Info("starting http server", "addr", cfg.ListenAddr)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
logger.Error("http server failed", "err", err)
cancel()
}
}()
logger.Info("startup complete")
<-ctx.Done()
logger.Info("received shutdown signal", "reason", ctx.Err())
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer shutdownCancel()
if err := srv.Shutdown(shutdownCtx); err != nil {
logger.Error("http shutdown", "err", err)
}
// ingest has to finish its transaction before the db closes
<-ingestDone
if err := database.Close(); err != nil {
logger.Error("db close", "err", err)
}
logger.Info("shutdown complete")
return nil
}
func chiMount(x *delxrpc.Xrpc) http.Handler {
mux := chi.NewRouter()
mux.Mount("/xrpc", x.Router())
return mux
}