Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386package sitesd
import ( "context" "encoding/hex" "encoding/json" "fmt" "log/slog" "net/http" "os" "strings" "time"
"github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/sitesd/cache" "tangled.org/core/sitesd/cloudflare" "tangled.org/core/sitesd/config" "tangled.org/core/sitesd/models" "tangled.org/core/sitesd/reposync" "tangled.org/core/sitesd/sitefeed" "tangled.org/core/sitesd/sites" "tangled.org/core/sitesd/worker")
const ( // deployTimeout bounds one deploy's R2 sync. deployTimeout = 10 * time.Minute // how long Stop lets in-flight deploys record their outcome deployShutdownGrace = 30 * time.Second // recordTimeout bounds the history write recordTimeout = 30 * time.Second // shutdownTimeout is the HTTP server drain budget at Stop. shutdownTimeout = 15 * time.Second preflightTimeout = 10 * time.Second internalWriteTimeout = 30 * time.Second)
// Options are the runtime parameters of the site deploy executor, provided by// the cmd/sitesd entrypoint (flags with env defaults).type Options struct { Listen string InternalListen string WorkerURL string Audience string Issuer string KeyHex string Knots string Jetstream string RedisAddr string Dev bool}
// Serve runs the sitesd daemon until ctx is canceled: the push feed,// the deploy HTTP surface (internal /deploy when InternalListen is set), did// document, health.func Serve(ctx context.Context, logger *slog.Logger, opts Options) error { var knots []string for _, knot := range strings.Split(opts.Knots, ",") { if knot = strings.TrimSpace(knot); knot != "" { knots = append(knots, knot) } } if opts.Jetstream == "" { return fmt.Errorf("no jetstream configured (SITESD_JETSTREAM_ENDPOINT); push-triggered deploys would never run") }
runCtx, cancel := context.WithCancel(ctx) defer cancel()
key := opts.KeyHex if key == "" { key = os.Getenv("SITESD_PRIVATE_KEY") } priv, err := loadKey(key) if err != nil { return fmt.Errorf("loading P-256 key: %w", err) }
cfg := &config.Config{ Core: config.Core{Dev: opts.Dev}, Cloudflare: config.Cloudflare{ AccountId: os.Getenv("CLOUDFLARE_ACCOUNT_ID"), R2: config.R2Config{ AccessKeyID: os.Getenv("CLOUDFLARE_R2_ACCESS_KEY_ID"), SecretAccessKey: os.Getenv("CLOUDFLARE_R2_SECRET_ACCESS_KEY"), Bucket: os.Getenv("CLOUDFLARE_R2_BUCKET"), BridgeURL: os.Getenv("SITESD_R2_BRIDGE_URL"), BridgeToken: os.Getenv("SITESD_R2_BRIDGE_TOKEN"), }, }, }
cf, err := cloudflare.New(cfg) if err != nil { return fmt.Errorf("building cloudflare client: %w", err) } if !opts.Dev && (!cf.Enabled() || cfg.Cloudflare.R2.AccessKeyID == "" || cfg.Cloudflare.R2.SecretAccessKey == "") { return fmt.Errorf("cloudflare R2 requires CLOUDFLARE_ACCOUNT_ID and CLOUDFLARE_R2_* credentials") } if !cf.Enabled() { logger.Warn("cloudflare is not configured (CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_R2_*): every deploy is refused") }
wrk := worker.New(opts.WorkerURL, opts.Issuer, opts.Audience, priv)
var rdb *cache.Cache if opts.RedisAddr != "" { rdb = cache.New(opts.RedisAddr) } else { logger.Warn("no redis address (SITESD_REDIS): the jetstream cursor is not persisted, so a restart resumes live and the pushes it missed never deploy") }
deployCtx, stopDeploys := context.WithCancel(context.Background()) defer stopDeploys()
locks := reposync.New() feed := sitefeed.New(rdb, cfg, cf, wrk, locks, deployCtx, knots, logger) if cf.Enabled() { if err := feed.Run(runCtx, opts.Jetstream); err != nil { return err } }
mux := http.NewServeMux() srv := &server{ cfg: cfg, cf: cf, wrk: wrk, locks: locks, logger: logger, deployCtx: deployCtx, getCtx: wrk.GetDeployContext, deploy: sites.Deploy, record: wrk.RecordDeploy, } mux.HandleFunc("GET /.well-known/did.json", srv.handleDidDoc(opts.Issuer, priv)) mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })
logger.Info("sitesd listening", "addr", opts.Listen, "worker", opts.WorkerURL, "did", opts.Issuer) httpServer := &http.Server{ Addr: opts.Listen, Handler: mux, ReadHeaderTimeout: 10 * time.Second, WriteTimeout: 10 * time.Minute, IdleTimeout: 2 * time.Minute, } serveErr := make(chan error, 2) go func() { if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { serveErr <- fmt.Errorf("public server: %w", err) } }()
var internalServer *http.Server if opts.InternalListen != "" { // private-network only: the worker cannot sign service auth internalMux := http.NewServeMux() internalMux.HandleFunc("POST /deploy", srv.handleInternalDeploy) internalServer = &http.Server{ Addr: opts.InternalListen, Handler: internalMux, ReadHeaderTimeout: 10 * time.Second, WriteTimeout: internalWriteTimeout, IdleTimeout: 2 * time.Minute, } go func() { logger.Info("internal deploy trigger listening", "addr", opts.InternalListen) if err := internalServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { serveErr <- fmt.Errorf("internal server: %w", err) } }() } else { logger.Warn("no internal listen address (SITESD_INTERNAL_ADDR): config-change and manual deploys have no endpoint") }
var serveErrVal error select { case err := <-serveErr: serveErrVal = err cancel() case <-runCtx.Done(): } logger.Info("shutting down", "reason", runCtx.Err()) shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), shutdownTimeout) defer shutdownCancel() if err := httpServer.Shutdown(shutdownCtx); err != nil { logger.Error("public server shutdown", "err", err) } if internalServer != nil { if err := internalServer.Shutdown(shutdownCtx); err != nil { logger.Error("internal server shutdown", "err", err) } } srv.drainDeploys(stopDeploys) return serveErrVal}
func (s *server) drainDeploys(stop context.CancelFunc) { done := make(chan struct{}) go func() { defer close(done) s.locks.Drain() }() select { case <-done: case <-time.After(deployShutdownGrace): s.logger.Warn("deploy drain grace elapsed, cancelling in-flight deploys", "grace", deployShutdownGrace) stop() <-done }}
func loadKey(hexStr string) (*atcrypto.PrivateKeyP256, error) { if hexStr == "" { return nil, fmt.Errorf("no key provided: set SITESD_PRIVATE_KEY or -key (64 hex chars)") } raw, err := hex.DecodeString(strings.TrimSpace(hexStr)) if err != nil { return nil, fmt.Errorf("decoding key hex: %w", err) } return atcrypto.ParsePrivateBytesP256(raw)}
type server struct { cfg *config.Config cf *cloudflare.Client wrk *worker.Client locks *reposync.Registry logger *slog.Logger
deployCtx context.Context
getCtx func(context.Context, string) (*worker.DeployContext, error) deploy func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error record func(context.Context, *worker.DeployContext, string, string, string, string) error}
type deployRequest struct { RepoDid string `json:"repoDid"` // an omitted trigger is a config-change deploy Trigger string `json:"trigger"`}
type acceptedDeploy struct { dc *worker.DeployContext trigger models.SiteDeployTrigger release func() done func()}
// ack 202 before deploy runs because worker wait_until expires in seconds while deploys take minutesfunc (s *server) handleInternalDeploy(w http.ResponseWriter, r *http.Request) { var req deployRequest if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&req); err != nil { writeErr(w, http.StatusBadRequest, "InvalidRequest", "request body must be JSON") return } if req.RepoDid == "" { writeErr(w, http.StatusBadRequest, "InvalidRequest", "missing repoDid") return } trigger := models.SiteDeployTriggerConfigChange if req.Trigger != "" { var ok bool if trigger, ok = models.ParseSiteDeployTrigger(req.Trigger); !ok { writeErr(w, http.StatusBadRequest, "InvalidRequest", "trigger must be one of config_change, push, manual") return } } if !s.cf.Enabled() { writeErr(w, http.StatusServiceUnavailable, "DeployUnavailable", "cloudflare is not configured") return }
preflightCtx, preflightCancel := context.WithTimeout(r.Context(), preflightTimeout) defer preflightCancel() dc, err := s.getCtx(preflightCtx, req.RepoDid) if err != nil { s.logger.Error("internal deploy: deploy context unavailable", "repo", req.RepoDid, "err", err) writeErr(w, http.StatusBadGateway, "DeployContextUnavailable", err.Error()) return } if dc == nil { writeErr(w, http.StatusUnprocessableEntity, "NoDeployContext", "no site config or active claim for this repo") return }
// register before ack so draining waits for accepted work done, ok := s.locks.Begin() if !ok { writeErr(w, http.StatusServiceUnavailable, "ShuttingDown", "deploy executor is shutting down") return } release, ok := s.locks.TryAcquire() if !ok { done() writeErr(w, http.StatusTooManyRequests, "DeployAtCapacity", "every deploy slot is busy") return }
go s.runDeploy(acceptedDeploy{dc: dc, trigger: trigger, release: release, done: done})
w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusAccepted) json.NewEncoder(w).Encode(map[string]string{"status": "accepted"})}
func (s *server) runDeploy(a acceptedDeploy) { defer a.done() defer a.release()
runCtx, cancel := context.WithTimeout(s.deployCtx, deployTimeout) defer cancel()
dc := a.dc deploy := &models.SiteDeploy{ RepoDid: syntax.DID(dc.RepoDid), Branch: dc.Branch, Dir: dc.Dir, CommitSHA: "", Trigger: a.trigger, } repo := &models.Repo{ Did: dc.OwnerDid, Name: dc.Name, Rkey: dc.Rkey, Knot: dc.Knot, RepoDid: dc.RepoDid, }
// push feed shares this lock, serializing against trigger deploys for the same repo unlock, lockErr := s.locks.Lock(runCtx, dc.RepoDid) var deployErr error if lockErr != nil { deployErr = fmt.Errorf("waiting for the repo lock: %w", lockErr) } else { defer unlock() deployErr = s.deploy(runCtx, s.cf, s.cfg, repo, dc.Branch, dc.Dir) } deploy.Status = models.SiteDeployStatusSuccess if deployErr != nil { deploy.Status = models.SiteDeployStatusFailure deploy.Error = deployErr.Error() s.logger.Error("internal deploy: failed", "repo", dc.RepoDid, "err", deployErr) }
// outside the deploy's budget: timeouts and shutdowns still record recordCtx, recordCancel := context.WithTimeout(context.Background(), recordTimeout) defer recordCancel() if err := s.record(recordCtx, dc, deploy.CommitSHA, string(deploy.Status), string(deploy.Trigger), deploy.Error); err != nil { s.logger.Error("internal deploy: recording the outcome failed", "repo", dc.RepoDid, "status", deploy.Status, "err", err) }}
func (s *server) handleDidDoc(issuer string, priv *atcrypto.PrivateKeyP256) http.HandlerFunc { pub, err := priv.PublicKey() multibase := "" if err == nil { multibase = pub.Multibase() } doc := map[string]any{ "id": issuer, "verificationMethod": []map[string]string{ { "id": issuer + "#atproto", "type": "Multikey", "controller": issuer, "publicKeyMultibase": multibase, }, }, } return func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(doc) }}
func writeErr(w http.ResponseWriter, status int, name, message string) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) json.NewEncoder(w).Encode(map[string]string{"error": name, "message": message})}