package 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 minutes
func (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})
}