diff --git a/appview/pipelines/logs.go b/appview/pipelines/logs.go index 7a4c78b2..172e501e 100644 --- a/appview/pipelines/logs.go +++ b/appview/pipelines/logs.go @@ -2,12 +2,14 @@ package pipelines import ( "html/template" + "path" "regexp" "strings" terminal "github.com/buildkite/terminal-to-html/v3" "github.com/gorilla/websocket" "tangled.org/core/appview/pages/markup" + "tangled.org/core/hostutil" ) // matches any ANSI escape sequence: ESC [ m @@ -79,10 +81,11 @@ func ReadLogs(conn *websocket.Conn, ch chan LogEvent) { } } -func SpindleURL(dev bool, spindle, knot, rkey, workflow string) string { - scheme := "wss" - if dev { - scheme = "ws" +func SpindleURL(spindle, knot, rkey, workflow string) string { + url, err := hostutil.EnsureWsScheme(spindle) + if err != nil { + return "" } - return scheme + "://" + strings.Join([]string{spindle, "logs", knot, rkey, workflow}, "/") + + return url + path.Join("/logs", knot, rkey, workflow) } diff --git a/appview/pipelines/pipelines.go b/appview/pipelines/pipelines.go index 76149a64..a22de03f 100644 --- a/appview/pipelines/pipelines.go +++ b/appview/pipelines/pipelines.go @@ -18,6 +18,7 @@ import ( "tangled.org/core/appview/pages" "tangled.org/core/appview/reporesolver" "tangled.org/core/eventconsumer" + "tangled.org/core/hostutil" "tangled.org/core/idresolver" "tangled.org/core/orm" "tangled.org/core/rbac" @@ -222,7 +223,11 @@ func (p *Pipelines) Logs(w http.ResponseWriter, r *http.Request) { return } - url := SpindleURL(p.config.Core.Dev, spindle, knot, rkey, workflow) + url := SpindleURL(spindle, knot, rkey, workflow) + if url == "" { + http.Error(w, "invalid spindle hostname", http.StatusBadRequest) + return + } l = l.With("url", url) clientConn, err := upgrader.Upgrade(w, r, nil) @@ -421,11 +426,17 @@ func (p *Pipelines) Cancel(w http.ResponseWriter, r *http.Request) { return } + hostname, noTLS, err := hostutil.ParseHostname(spindle) + if err != nil { + http.Error(w, "invalid spindle hostname", http.StatusBadRequest) + return + } + spindleClient, err := p.oauth.ServiceClient( r, - oauth.WithService(f.Spindle), + oauth.WithService(hostname), oauth.WithLxm(tangled.PipelineCancelPipelineNSID), - oauth.WithDev(p.config.Core.Dev), + oauth.WithDev(noTLS), oauth.WithTimeout(time.Second*30), // workflow cleanup usually takes time ) diff --git a/appview/pipelines/ssh/tui.go b/appview/pipelines/ssh/tui.go index 01cc44cd..66f7d9d6 100644 --- a/appview/pipelines/ssh/tui.go +++ b/appview/pipelines/ssh/tui.go @@ -113,7 +113,7 @@ func (m *pipelineModel) connectCmd(workflow string) tea.Cmd { if !ok || len(ws.Data) == 0 { return logDoneMsg{workflow: workflow} } - url := pipelines.SpindleURL(m.server.config.Core.Dev, ws.Data[0].Spindle, m.pipeline.Knot, m.pipeline.Rkey, workflow) + url := pipelines.SpindleURL(ws.Data[0].Spindle, m.pipeline.Knot, m.pipeline.Rkey, workflow) conn, _, err := websocket.DefaultDialer.Dial(url, nil) if err != nil { return logDoneMsg{workflow: workflow, err: fmt.Errorf("connecting to spindle: %w", err)} diff --git a/appview/state/knotstream.go b/appview/state/knotstream.go index 17352dec..05e0877a 100644 --- a/appview/state/knotstream.go +++ b/appview/state/knotstream.go @@ -46,7 +46,7 @@ func Knotstream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac. return bootstrapStream( ctx, "knotstream", ec.KindKnot, hosts, c.Redis.Addr, - c.Knotstream, c.Core.Dev, + c.Knotstream, knotIngester(d, enforcer, posthog, notifier, c.Core.Dev, c, cfClient), ), nil } diff --git a/appview/state/spindlestream.go b/appview/state/spindlestream.go index ebe2643a..6cfd5aba 100644 --- a/appview/state/spindlestream.go +++ b/appview/state/spindlestream.go @@ -33,7 +33,7 @@ func Spindlestream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rb return bootstrapStream( ctx, "spindlestream", ec.KindSpindle, hosts, c.Redis.Addr, - c.Spindlestream, c.Core.Dev, + c.Spindlestream, spindleIngester(d, pn), ), nil } diff --git a/appview/state/spindlestream_test.go b/appview/state/spindlestream_test.go index 05cccf57..8b515a19 100644 --- a/appview/state/spindlestream_test.go +++ b/appview/state/spindlestream_test.go @@ -55,7 +55,7 @@ func TestColdStart_SpindleEventsRebuildPipelineStatuses(t *testing.T) { }) srv := httptest.NewServer(mux) t.Cleanup(srv.Close) - source := ec.Source{Kind: "test", Host: strings.TrimPrefix(srv.URL, "http://")} + source := ec.Source{Kind: "test", Host: strings.TrimPrefix(srv.URL, "http://"), NoTLS: true} appviewDB, err := db.Make(ctx, filepath.Join(t.TempDir(), "appview.db")) if err != nil { @@ -72,7 +72,6 @@ func TestColdStart_SpindleEventsRebuildPipelineStatuses(t *testing.T) { QueueSize: 16, ConnectionTimeout: 2 * time.Second, CursorStore: &cursor.MemoryStore{}, - URLFunc: ec.DefaultURL(true), Logger: logger, } c := ec.NewConsumer(cfg) diff --git a/appview/state/streams.go b/appview/state/streams.go index 59fabf68..4e7c748a 100644 --- a/appview/state/streams.go +++ b/appview/state/streams.go @@ -17,7 +17,6 @@ func bootstrapStream( hosts []string, redisAddr string, streamCfg config.ConsumerConfig, - dev bool, processFn ec.ProcessFunc, ) *ec.Consumer { logger := log.SubLogger(log.FromContext(ctx), name) @@ -41,7 +40,6 @@ func bootstrapStream( WorkerCount: streamCfg.WorkerCount, QueueSize: streamCfg.QueueSize, Logger: logger, - URLFunc: ec.DefaultURL(dev), CursorStore: &cursorStore, }) } diff --git a/eventconsumer/consumer.go b/eventconsumer/consumer.go index b540f989..08182f5a 100644 --- a/eventconsumer/consumer.go +++ b/eventconsumer/consumer.go @@ -5,7 +5,6 @@ import ( "encoding/json" "log/slog" "net/http" - "net/url" "sync" "time" @@ -29,7 +28,6 @@ type ConsumerConfig struct { QueueSize int Logger *slog.Logger CursorStore cursor.Store - URLFunc func(Source, int64) (*url.URL, error) Dialer *websocket.Dialer RequestHeader http.Header @@ -93,9 +91,6 @@ func NewConsumer(cfg ConsumerConfig) *Consumer { if cfg.CursorStore == nil { cfg.CursorStore = &cursor.MemoryStore{} } - if cfg.URLFunc == nil { - cfg.URLFunc = DefaultURL(false) - } dialer := cfg.Dialer if dialer == nil { dialer = websocket.DefaultDialer @@ -263,7 +258,7 @@ func (c *Consumer) startConnectionLoop(ctx context.Context, source Source) { func (c *Consumer) runConnection(ctx context.Context, source Source) error { cursor := c.cfg.CursorStore.Get(source.Key()) - u, err := c.cfg.URLFunc(source, cursor) + u, err := source.URL(cursor) if err != nil { return err } diff --git a/eventconsumer/consumer_test.go b/eventconsumer/consumer_test.go index d4255234..2cd19082 100644 --- a/eventconsumer/consumer_test.go +++ b/eventconsumer/consumer_test.go @@ -69,7 +69,7 @@ func startEventServer(t *testing.T, src *memSrc) (Source, *notifier.Notifier) { srv := httptest.NewServer(mux) t.Cleanup(srv.Close) addr := strings.TrimPrefix(srv.URL, "http://") - return Source{Kind: "test", Host: addr}, &n + return Source{Kind: "test", Host: addr, NoTLS: true}, &n } func TestConsumer_DrainAdvancesCursor(t *testing.T) { @@ -95,7 +95,6 @@ func TestConsumer_DrainAdvancesCursor(t *testing.T) { QueueSize: 16, ConnectionTimeout: 2 * time.Second, CursorStore: store, - URLFunc: DefaultURL(true), Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), } c := NewConsumer(cfg) @@ -158,7 +157,6 @@ func TestConsumer_CursorMonotonic_OutOfOrderWorkers(t *testing.T) { QueueSize: 16, ConnectionTimeout: 2 * time.Second, CursorStore: store, - URLFunc: DefaultURL(true), Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), } c := NewConsumer(cfg) @@ -203,7 +201,6 @@ func TestConsumer_StopTerminatesWithoutCtxCancel(t *testing.T) { QueueSize: 8, ConnectionTimeout: 2 * time.Second, CursorStore: &cursor.MemoryStore{}, - URLFunc: DefaultURL(true), Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), } c := NewConsumer(cfg) @@ -249,7 +246,6 @@ func TestConsumer_ResumesFromStoredCursor(t *testing.T) { QueueSize: 16, ConnectionTimeout: 2 * time.Second, CursorStore: store, - URLFunc: DefaultURL(true), Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), } c := NewConsumer(cfg) diff --git a/eventconsumer/source.go b/eventconsumer/source.go index f0b768cb..010d4267 100644 --- a/eventconsumer/source.go +++ b/eventconsumer/source.go @@ -5,6 +5,7 @@ import ( "strconv" "tangled.org/core/eventconsumer/cursor" + "tangled.org/core/hostutil" ) type Kind string @@ -15,12 +16,19 @@ const ( ) type Source struct { - Kind Kind - Host string + Kind Kind + Host string + NoTLS bool // use TLS by default } -func NewKnotSource(host string) Source { return Source{Kind: KindKnot, Host: host} } -func NewSpindleSource(host string) Source { return Source{Kind: KindSpindle, Host: host} } +func NewKnotSource(host string) Source { + host, noTLS, _ := hostutil.ParseHostname(host) + return Source{Kind: KindKnot, Host: host, NoTLS: noTLS} +} +func NewSpindleSource(host string) Source { + host, noTLS, _ := hostutil.ParseHostname(host) + return Source{Kind: KindSpindle, Host: host, NoTLS: noTLS} +} func (s Source) Key() string { return string(s.Kind) + ":" + s.Host } @@ -33,21 +41,19 @@ func MigrateLegacyCursor(store cursor.Store, s Source) { } } -func DefaultURL(dev bool) func(Source, int64) (*url.URL, error) { +func (s Source) URL(cursor int64) (*url.URL, error) { scheme := "wss" - if dev { + if s.NoTLS { scheme = "ws" } - return func(s Source, cursor int64) (*url.URL, error) { - u, err := url.Parse(scheme + "://" + s.Host + "/events") - if err != nil { - return nil, err - } - if cursor != 0 { - q := url.Values{} - q.Add("cursor", strconv.FormatInt(cursor, 10)) - u.RawQuery = q.Encode() - } - return u, nil + u, err := url.Parse(scheme + "://" + s.Host + "/events") + if err != nil { + return nil, err + } + if cursor != 0 { + q := url.Values{} + q.Add("cursor", strconv.FormatInt(cursor, 10)) + u.RawQuery = q.Encode() } + return u, nil } diff --git a/eventconsumer/upgrade_test.go b/eventconsumer/upgrade_test.go index 7e92425e..3c90f34b 100644 --- a/eventconsumer/upgrade_test.go +++ b/eventconsumer/upgrade_test.go @@ -39,7 +39,6 @@ func drainProcessed(t *testing.T, store cursor.Store, source Source) []int64 { QueueSize: 16, ConnectionTimeout: 2 * time.Second, CursorStore: store, - URLFunc: DefaultURL(true), Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), }) diff --git a/knotmirror/hostutil/hostutil.go b/hostutil/hostutil.go similarity index 74% rename from knotmirror/hostutil/hostutil.go rename to hostutil/hostutil.go index 26c9b2b7..2038a06b 100644 --- a/knotmirror/hostutil/hostutil.go +++ b/hostutil/hostutil.go @@ -54,3 +54,27 @@ func ParseHostname(raw string) (hostname string, noSSL bool, err error) { // lower-case in response return h.Normalize().String(), noSSL, nil } + +func EnsureHttpScheme(host string) (string, error) { + hostname, noSSL, err := ParseHostname(host) + if err != nil { + return "", err + } + if noSSL { + return "http://" + hostname, nil + } else { + return "https://" + hostname, nil + } +} + +func EnsureWsScheme(host string) (string, error) { + hostname, noSSL, err := ParseHostname(host) + if err != nil { + return "", err + } + if noSSL { + return "ws://" + hostname, nil + } else { + return "wss://" + hostname, nil + } +} diff --git a/knotmirror/xrpc/sync_request_crawl.go b/knotmirror/xrpc/sync_request_crawl.go index 900b8beb..5858ec38 100644 --- a/knotmirror/xrpc/sync_request_crawl.go +++ b/knotmirror/xrpc/sync_request_crawl.go @@ -11,8 +11,8 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/xrpc" "tangled.org/core/api/tangled" + "tangled.org/core/hostutil" "tangled.org/core/knotmirror/db" - "tangled.org/core/knotmirror/hostutil" "tangled.org/core/knotmirror/models" ) diff --git a/spindle/models/clone.go b/spindle/models/clone.go index d3a17574..b5219795 100644 --- a/spindle/models/clone.go +++ b/spindle/models/clone.go @@ -5,6 +5,7 @@ import ( "strings" "tangled.org/core/api/tangled" + "tangled.org/core/hostutil" "tangled.org/core/workflow" ) @@ -63,13 +64,20 @@ func BuildCloneStep(twf tangled.Pipeline_Workflow, tr tangled.Pipeline_TriggerMe } fetchArgs := buildFetchArgs(cloneOpts, commitSHA) + // In dev mode we point at Caddy via host-gateway with a self-signed cert, + // so skip the TLS check for the fetch call. + fetchCmd := "git fetch" + if dev { + fetchCmd = "git -c http.sslVerify=false fetch" + } + return CloneStep{ kind: StepKindSystem, name: "Clone repository into workspace", commands: []string{ "git init", fmt.Sprintf("git remote add origin %s", repoURL), - fmt.Sprintf("git fetch %s", strings.Join(fetchArgs, " ")), + fmt.Sprintf("%s %s", fetchCmd, strings.Join(fetchArgs, " ")), "git checkout FETCH_HEAD", }, } @@ -103,20 +111,18 @@ func extractCommitSHA(tr tangled.Pipeline_TriggerMetadata) (string, error) { // BuildRepoURL constructs the repository URL from repo metadata. func BuildRepoURL(repo *tangled.Pipeline_TriggerRepo, devMode bool) string { - scheme := "https://" - if devMode { - scheme = "http://" + host, noSSL, _ := hostutil.ParseHostname(repo.Knot) + scheme := "https" + if noSSL { + scheme = "http" } - // Get host from knot - host := repo.Knot - // In dev mode, replace localhost with host.docker.internal for Docker networking - if devMode && strings.Contains(host, "localhost") { + if devMode && strings.HasPrefix(host, "localhost:") { host = strings.ReplaceAll(host, "localhost", "host.docker.internal") } - return fmt.Sprintf("%s%s/%s", scheme, host, *repo.RepoDid) + return fmt.Sprintf("%s://%s/%s", scheme, host, *repo.RepoDid) } // buildFetchArgs constructs the arguments for git fetch based on clone options diff --git a/spindle/server.go b/spindle/server.go index 4648c663..a661754d 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -185,7 +185,6 @@ func New(ctx context.Context, cfg *config.Config, engines map[string]models.Engi // job in the above registered queue. ccfg := eventconsumer.NewConsumerConfig() ccfg.Logger = log.SubLogger(logger, "eventconsumer") - ccfg.URLFunc = eventconsumer.DefaultURL(cfg.Server.Dev) ccfg.ProcessFunc = spindle.processPipeline ccfg.CursorStore = cursorStore knownKnots, err := d.Knots()