Stitch any CI into Tangled
1// Tack is a custom Tangled spindle that translates sh.tangled.pipeline
2// trigger records into Buildkite builds, and publishes Buildkite job state
3// back as sh.tangled.pipeline.status events on a WebSocket stream that the
4// Tangled appview can consume.
5package main
6
7import (
8 "context"
9 "errors"
10 "flag"
11 "fmt"
12 "log/slog"
13 "os"
14 "os/signal"
15 "syscall"
16
17 charmlog "github.com/charmbracelet/log"
18
19 "github.com/mitchellh/tack/internal/buildkite"
20)
21
22// config is the runtime configuration, sourced from environment variables and
23// flags. Env vars match the README so this can be swapped in for the stock
24// spindle without surprises.
25type config struct {
26 Addr string
27 Hostname string
28 OwnerDID string
29 JetstreamURL string
30 DBPath string
31 // Dev flips the knot event-stream scheme from wss:// to ws://.
32 // Useful when running against a local knot during development.
33 Dev bool
34
35 // Buildkite-mode configuration. BuildkiteToken is the switch:
36 // when empty we fall back to the in-process fake provider
37 // (useful for local development against a real Tangled
38 // jetstream); when set, the other Buildkite fields are
39 // required and tack will refuse to start without them.
40 BuildkiteToken string
41 BuildkiteOrg string
42 BuildkitePipeline string
43 BuildkiteWebhookSecret string
44 BuildkiteWebhookMode buildkite.WebhookMode
45}
46
47func loadConfig() (config, error) {
48 cfg := config{
49 Addr: envOr("TACK_LISTEN_ADDR", ":8080"),
50 Hostname: os.Getenv("TACK_HOSTNAME"),
51 OwnerDID: os.Getenv("TACK_OWNER_DID"),
52 JetstreamURL: envOr("TACK_JETSTREAM_URL", "wss://jetstream1.us-west.bsky.network/subscribe"),
53 DBPath: envOr("TACK_DB_PATH", "tack.db"),
54 Dev: os.Getenv("TACK_DEV") != "",
55 BuildkiteToken: os.Getenv("TACK_BUILDKITE_TOKEN"),
56 BuildkiteOrg: os.Getenv("TACK_BUILDKITE_ORG"),
57 BuildkitePipeline: os.Getenv("TACK_BUILDKITE_PIPELINE"),
58 BuildkiteWebhookSecret: os.Getenv("TACK_BUILDKITE_WEBHOOK_SECRET"),
59 BuildkiteWebhookMode: buildkite.WebhookMode(
60 envOr("TACK_BUILDKITE_WEBHOOK_MODE", string(buildkite.WebhookModeToken)),
61 ),
62 }
63 addrFlag := flag.String("addr", cfg.Addr, "HTTP listen address (overrides TACK_LISTEN_ADDR)")
64 flag.Parse()
65 cfg.Addr = *addrFlag
66
67 if cfg.OwnerDID == "" {
68 return cfg, errors.New("TACK_OWNER_DID is required")
69 }
70
71 // Hostname identifies *us* in sh.tangled.repo records (the .spindle
72 // field). Without it we have no way to know which repos point at us
73 // and therefore which knots we should subscribe to for pipeline
74 // triggers — so we refuse to start rather than silently subscribe to
75 // nothing.
76 if cfg.Hostname == "" {
77 return cfg, errors.New("TACK_HOSTNAME is required")
78 }
79
80 // If the operator opted into Buildkite mode (by supplying a
81 // token), every other Buildkite knob has to be present. Half-
82 // configured Buildkite leads to confusing failures deep in the
83 // provider; catch it at startup.
84 if cfg.BuildkiteToken != "" {
85 if cfg.BuildkiteOrg == "" {
86 return cfg, errors.New("TACK_BUILDKITE_ORG is required when TACK_BUILDKITE_TOKEN is set")
87 }
88 if cfg.BuildkitePipeline == "" {
89 return cfg, errors.New("TACK_BUILDKITE_PIPELINE is required when TACK_BUILDKITE_TOKEN is set")
90 }
91 if cfg.BuildkiteWebhookSecret == "" {
92 return cfg, errors.New("TACK_BUILDKITE_WEBHOOK_SECRET is required when TACK_BUILDKITE_TOKEN is set")
93 }
94 switch cfg.BuildkiteWebhookMode {
95 case buildkite.WebhookModeToken, buildkite.WebhookModeSignature:
96 default:
97 return cfg, fmt.Errorf("TACK_BUILDKITE_WEBHOOK_MODE must be %q or %q; got %q",
98 buildkite.WebhookModeToken, buildkite.WebhookModeSignature,
99 cfg.BuildkiteWebhookMode,
100 )
101 }
102 }
103
104 return cfg, nil
105}
106
107func envOr(key, def string) string {
108 if v := os.Getenv(key); v != "" {
109 return v
110 }
111 return def
112}
113
114func main() {
115 // Logging setup. charmbracelet/log implements slog.Handler, so we wrap
116 // it in slog.New to share the same backend with libraries that expect
117 // a *slog.Logger (notably the jetstream client).
118 charmHandler := charmlog.NewWithOptions(os.Stderr, charmlog.Options{
119 Level: charmlog.DebugLevel,
120 ReportTimestamp: true,
121 })
122 logger := slog.New(charmHandler)
123 slog.SetDefault(logger)
124
125 // Config loading
126 cfg, err := loadConfig()
127 if err != nil {
128 logger.Error("invalid configuration", "err", err)
129 os.Exit(2)
130 }
131
132 // Root context: cancelled on SIGINT/SIGTERM, with the logger attached
133 // so any function we hand it to can pull it back out via loggerFrom.
134 ctx, stop := signal.NotifyContext(
135 context.Background(),
136 os.Interrupt, syscall.SIGTERM,
137 )
138 defer stop()
139 ctx = loggerInto(ctx, logger)
140
141 // Open (or create) the SQLite store. Holds jetstream cursor +
142 // observed Tangled membership records. Closed last during shutdown
143 // so anything writing to it during teardown still succeeds.
144 st, err := openStore(cfg.DBPath)
145 if err != nil {
146 logger.Error("failed to open store", "err", err, "path", cfg.DBPath)
147 os.Exit(1)
148 }
149 defer func() {
150 if err := st.Close(); err != nil {
151 logger.Error("close store", "err", err)
152 }
153 }()
154 logger.Info("store open", "path", cfg.DBPath)
155
156 // In-process broker for the /events fan-out. Wraps the store so
157 // publishes are durable and reconnecting subscribers can resume by
158 // cursor. Constructed before the consumers in case we ever want
159 // them to publish synthetic status events at startup.
160 br := newBroker(st)
161
162 // Provider that turns Tangled pipeline triggers into
163 // pipeline.status events. The Buildkite provider is the real
164 // integration; the fake one stands in when no Buildkite token is
165 // configured so the full jetstream → knot → /events flow is
166 // still exercisable locally without a Buildkite account.
167 //
168 // bkProvider is kept as a typed pointer separately because the
169 // /webhooks/buildkite handler needs the concrete *buildkiteProvider
170 // (for HandleWebhook + signature verification), not the abstract
171 // Provider surface.
172 var (
173 provider Provider
174 bkProvider *buildkiteProvider
175 )
176 if cfg.BuildkiteToken != "" {
177 bkProvider = newBuildkiteProvider(
178 br, st,
179 buildkite.NewClient(cfg.BuildkiteToken, cfg.BuildkiteOrg),
180 cfg.BuildkitePipeline,
181 cfg.BuildkiteWebhookSecret,
182 cfg.BuildkiteWebhookMode,
183 logger,
184 )
185 provider = bkProvider
186 logger.Info("buildkite provider enabled",
187 "org", cfg.BuildkiteOrg,
188 "pipeline", cfg.BuildkitePipeline,
189 "webhook_mode", cfg.BuildkiteWebhookMode,
190 )
191 } else {
192 provider = newFakeProvider(br, logger)
193 logger.Info("fake provider enabled (set TACK_BUILDKITE_TOKEN to use buildkite)")
194 }
195
196 // Start the knot event-stream consumer first so the jetstream
197 // loop has somewhere to register newly-observed knots into. It
198 // gets the provider so each incoming pipeline trigger has
199 // something to dispatch to.
200 knots, err := startKnotConsumer(ctx, cfg, st, provider)
201 if err != nil {
202 logger.Error("failed to start knot consumer", "err", err)
203 os.Exit(1)
204 }
205 defer knots.Stop()
206
207 // Start the JetStream listener in the background. It hands the knot
208 // consumer any new knot referenced by an incoming sh.tangled.repo
209 // record so we don't have to wait for a restart to pick it up.
210 if err := startJetstream(ctx, cfg, st, knots); err != nil {
211 logger.Error("failed to start jetstream consumer", "err", err)
212 os.Exit(1)
213 }
214
215 // Run the HTTP server. This blocks until ctx is cancelled or the
216 // listener errors.
217 if err := runHTTP(ctx, cfg, br, provider, bkProvider); err != nil {
218 logger.Error("http server error", "err", err)
219 os.Exit(1)
220 }
221}