pass callback for logging in

cmd/humanlog/localhost.go

@@ -12,10 +12,8 @@ import (
 	"sync"
 	"time"
 
-	"connectrpc.com/connect"
 	connectcors "connectrpc.com/cors"
 
-	"github.com/humanlogio/api/go/svc/feature/v1/featurev1connect"
 	"github.com/humanlogio/api/go/svc/ingest/v1/ingestv1connect"
 	"github.com/humanlogio/api/go/svc/localhost/v1/localhostv1connect"
 	"github.com/humanlogio/api/go/svc/query/v1/queryv1connect"
@@ -45,10 +43,8 @@ func startLocalhostServer(
 	machineID uint64,
 	port int,
 	localhostHttpClient *http.Client,
-	apiHttpClient *http.Client,
-	apiURL string,
-	apiClientOpts []connect.ClientOption,
 	ownVersion *typesv1.Version,
+	app *localstorage.AppCtx,
 ) (localsink sink.Sink, done func(context.Context) error, err error) {
 
 	notifyUnableToIngest := func(err error) {
@@ -79,14 +75,12 @@ func startLocalhostServer(
 		}, nil
 	}
 
-	features := featurev1connect.NewFeatureServiceClient(apiHttpClient, apiURL, apiClientOpts...)
-
 	storage, err := localstorage.Open(
 		ctx,
 		cfg.ExperimentalFeatures.ServeLocalhost.Engine,
 		ll.WithGroup("storage"),
 		cfg.ExperimentalFeatures.ServeLocalhost.Cfg,
-		features,
+		app,
 	)
 	if err != nil {
 		return nil, nil, fmt.Errorf("opening localstorage %q: %v", cfg.ExperimentalFeatures.ServeLocalhost.Engine, err)

cmd/humanlog/main.go

@@ -22,11 +22,13 @@ import (
 	"github.com/blang/semver"
 	"github.com/charmbracelet/huh"
 	"github.com/gen2brain/beeep"
+	"github.com/humanlogio/api/go/svc/feature/v1/featurev1connect"
 	types "github.com/humanlogio/api/go/types/v1"
 	"github.com/humanlogio/humanlog"
 	"github.com/humanlogio/humanlog/internal/pkg/config"
 	"github.com/humanlogio/humanlog/internal/pkg/state"
 	"github.com/humanlogio/humanlog/pkg/auth"
+	"github.com/humanlogio/humanlog/pkg/localstorage"
 	"github.com/humanlogio/humanlog/pkg/sink"
 	"github.com/humanlogio/humanlog/pkg/sink/stdiosink"
 	"github.com/humanlogio/humanlog/pkg/sink/teesink"
@@ -233,14 +235,19 @@ func newApp() *cli.App {
 		cancel     context.CancelFunc
 		cfg        *config.Config
 		statefile  *state.State
+		dialer     = &net.Dialer{Timeout: time.Second}
 		httpClient = &http.Client{
-			Transport: &http2.Transport{},
+			Transport: &http2.Transport{
+				DialTLS: func(network, addr string, cfg *tls.Config) (net.Conn, error) {
+					return tls.DialWithDialer(dialer, network, addr, cfg)
+				},
+			},
 		}
 		localhostHttpClient = &http.Client{
 			Transport: &http2.Transport{
 				AllowHTTP: true,
 				DialTLS: func(network, addr string, _ *tls.Config) (net.Conn, error) {
-					return net.Dial(network, addr)
+					return dialer.Dial(network, addr)
 				},
 			},
 		}
@@ -533,19 +540,26 @@ func newApp() *cli.App {
 						return fmt.Errorf("this feature requires a valid machine ID, which requires an environment. failed to login: %v", err)
 					}
 				}
-				clOpts := []connect.ClientOption{connect.WithInterceptors(
-					auth.Interceptors(ll, getTokenSource(cctx))...,
-				)}
+				apiHttpClient := getHTTPClient(cctx, getAPIUrl(cctx))
+				apiClientOpts := []connect.ClientOption{
+					connect.WithInterceptors(auth.Interceptors(ll, getTokenSource(cctx))...),
+				}
 
 				machineID = uint64(*state.MachineID)
 				localhostSink, done, err := startLocalhostServer(
 					ctx, ll, cfg, state, machineID, localhostCfg.Port,
 					getLocalhostHTTPClient(cctx),
-					getHTTPClient(cctx, getAPIUrl(cctx)), getAPIUrl(cctx), clOpts,
 					version,
+					&localstorage.AppCtx{
+						EnsureLoggedIn: func(ctx context.Context) error {
+							_, err := ensureLoggedIn(ctx, cctx, state, getTokenSource(cctx), apiURL, getHTTPClient(cctx, apiURL))
+							return err
+						},
+						Features: featurev1connect.NewFeatureServiceClient(apiHttpClient, apiURL, apiClientOpts...),
+					},
 				)
 				if err != nil {
-					loginfo("starting experimental localhost service: %v", err)
+					logerror("failed to start localhost service: %v", err)
 				} else {
 					sink = teesink.NewTeeSink(sink, localhostSink)
 					defer func() {

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/go-logfmt/logfmt v0.5.1
 	github.com/google/uuid v1.6.0
 	github.com/humanlogio/api/go v0.0.0-20241128170213-590d167300cd
-	github.com/humanlogio/humanlog-pro v0.0.0-20241129054503-5f8dc005f887
+	github.com/humanlogio/humanlog-pro v0.0.0-20241129104046-2603398b2ef1
 	github.com/kr/logfmt v0.0.0-20210122060352-19f9bcb100e6
 	github.com/lrstanley/bubblezone v0.0.0-20240914071701-b48c55a5e78e
 	github.com/matoous/go-nanoid v1.5.0

go.sum

@@ -82,8 +82,8 @@ github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c h1:6rhixN/i8
 github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c/go.mod h1:NMPJylDgVpX0MLRlPy15sqSwOFv/U1GZ2m21JhFfek0=
 github.com/humanlogio/api/go v0.0.0-20241128170213-590d167300cd h1:449C6cnB4W6DblDMPfCfA4xyEkiYMpngGf7TEX9O8ro=
 github.com/humanlogio/api/go v0.0.0-20241128170213-590d167300cd/go.mod h1:+hU/MU1g6QvtbeknKOlUI1yEStVqkPJ8jmYIj63OV5I=
-github.com/humanlogio/humanlog-pro v0.0.0-20241129054503-5f8dc005f887 h1:hUNsdeg5gm5s8GqJUTq3qs61anWMB1af1zL0+K3nIeE=
-github.com/humanlogio/humanlog-pro v0.0.0-20241129054503-5f8dc005f887/go.mod h1:H9mp76M+tzWQdACGpgyXWK/kwew7FLRtWzpY0pNWN4M=
+github.com/humanlogio/humanlog-pro v0.0.0-20241129104046-2603398b2ef1 h1:eDsZPPr9GBNNh8wy43wPlb/98R1uCHuxML/r99GIkP4=
+github.com/humanlogio/humanlog-pro v0.0.0-20241129104046-2603398b2ef1/go.mod h1:q+Kj7V58BQVVzGgcoyhtE2RXXprEi6DrqmhV/cThKkc=
 github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
 github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/klauspost/cpuid/v2 v2.2.8 h1:+StwCXwm9PdpiEkPyzBXIy+M9KUb4ODm0Zarf1kS5BM=

internal/memstorage/memory.go

@@ -8,7 +8,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/humanlogio/api/go/svc/feature/v1/featurev1connect"
 	typesv1 "github.com/humanlogio/api/go/types/v1"
 	"github.com/humanlogio/humanlog/pkg/localstorage"
 	"github.com/humanlogio/humanlog/pkg/sink"
@@ -24,7 +23,7 @@ var (
 )
 
 func init() {
-	localstorage.RegisterStorage("basic", func(ctx context.Context, ll *slog.Logger, cfg map[string]interface{}, features featurev1connect.FeatureServiceClient) (localstorage.Storage, error) {
+	localstorage.RegisterStorage("basic", func(ctx context.Context, ll *slog.Logger, cfg map[string]interface{}, appctx *localstorage.AppCtx) (localstorage.Storage, error) {
 		return NewMemStorage(ll), nil
 	})
 }

pkg/localstorage/queryable.go

@@ -11,11 +11,16 @@ import (
 	"github.com/humanlogio/humanlog/pkg/sink"
 )
 
+type AppCtx struct {
+	EnsureLoggedIn func(ctx context.Context) error
+	Features       featurev1connect.FeatureServiceClient
+}
+
 type StorageBuilder func(
 	ctx context.Context,
 	ll *slog.Logger,
 	cfg map[string]interface{},
-	features featurev1connect.FeatureServiceClient,
+	app *AppCtx,
 ) (Storage, error)
 
 var registry = make(map[string]StorageBuilder)
@@ -28,12 +33,12 @@ func RegisterStorage(name string, builder StorageBuilder) {
 	registry[name] = builder
 }
 
-func Open(ctx context.Context, name string, ll *slog.Logger, cfg map[string]interface{}, features featurev1connect.FeatureServiceClient) (Storage, error) {
+func Open(ctx context.Context, name string, ll *slog.Logger, cfg map[string]interface{}, app *AppCtx) (Storage, error) {
 	builder, ok := registry[name]
 	if !ok {
 		return nil, fmt.Errorf("no storage engine with name %q", name)
 	}
-	return builder(ctx, ll, cfg, features)
+	return builder(ctx, ll, cfg, app)
 }
 
 type Storage interface {