package sitesd
import (
"context"
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"tangled.org/core/sitesd/cloudflare"
"tangled.org/core/sitesd/config"
"tangled.org/core/sitesd/models"
"tangled.org/core/sitesd/reposync"
"tangled.org/core/sitesd/worker"
)
const (
testOwnerDid = "did:plc:mockowner"
testRepoDid = "did:plc:repo1"
testKnot = "knot.invalid"
waitFor = 5 * time.Second
)
var errBoom = errors.New("boom")
func TestServeRequiresR2OutsideDev(t *testing.T) {
for _, tc := range []struct {
name, account, bucket, access, secret string
}{
{name: "account", bucket: "bucket", access: "access", secret: "secret"},
{name: "bucket", account: "account", access: "access", secret: "secret"},
{name: "access", account: "account", bucket: "bucket", secret: "secret"},
{name: "secret", account: "account", bucket: "bucket", access: "access"},
} {
t.Run(tc.name, func(t *testing.T) {
t.Setenv("CLOUDFLARE_ACCOUNT_ID", tc.account)
t.Setenv("CLOUDFLARE_R2_BUCKET", tc.bucket)
t.Setenv("CLOUDFLARE_R2_ACCESS_KEY_ID", tc.access)
t.Setenv("CLOUDFLARE_R2_SECRET_ACCESS_KEY", tc.secret)
opts := Options{Knots: testKnot, Jetstream: "wss://jetstream.invalid/subscribe", KeyHex: strings.Repeat("0", 63) + "1"}
err := Serve(context.Background(), slog.New(slog.NewTextHandler(io.Discard, nil)), opts)
if err == nil || !strings.Contains(err.Error(), "cloudflare R2 requires") {
t.Fatalf("Serve with missing %s: %v", tc.name, err)
}
})
}
}
type serverHooks struct {
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
}
func newTestServer(hooks serverHooks) *server {
cf, err := cloudflare.New(&config.Config{Cloudflare: config.Cloudflare{
AccountId: "acct",
R2: config.R2Config{Bucket: "bucket"},
}})
if err != nil {
panic(err)
}
return &server{
cfg: &config.Config{},
cf: cf,
locks: reposync.New(),
logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
deployCtx: context.Background(),
getCtx: hooks.getCtx,
deploy: hooks.deploy,
record: hooks.record,
}
}
func internalDeployRequest(body string) *http.Request {
return httptest.NewRequest(http.MethodPost, "/deploy", strings.NewReader(body))
}
func ctxFor(overrides ...func(*worker.DeployContext)) *worker.DeployContext {
dc := &worker.DeployContext{
RepoDid: testRepoDid,
OwnerDid: testOwnerDid,
Name: "myblog",
Rkey: "abc1",
Knot: testKnot,
Branch: "main",
Dir: "/",
}
for _, o := range overrides {
o(dc)
}
return dc
}
func decodeStatus(t *testing.T, w *httptest.ResponseRecorder) string {
t.Helper()
var out struct {
Status string `json:"status"`
}
if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil {
t.Fatalf("decoding body %q: %v", w.Body.String(), err)
}
return out.Status
}
func await(t *testing.T, ch <-chan struct{}, what string) {
t.Helper()
select {
case <-ch:
case <-time.After(waitFor):
t.Fatalf("timed out waiting for %s", what)
}
}
func awaitQuiet(t *testing.T, ch <-chan struct{}, d time.Duration, why string) {
t.Helper()
select {
case <-ch:
t.Fatal(why)
case <-time.After(d):
}
}
func requireAccepted(t *testing.T, w *httptest.ResponseRecorder) {
t.Helper()
if w.Code != http.StatusAccepted {
t.Fatalf("status = %d, want 202 (body %q)", w.Code, w.Body.String())
}
if got := decodeStatus(t, w); got != "accepted" {
t.Errorf("status body = %q, want accepted", got)
}
}
func requireRefused(t *testing.T, srv *server, w *httptest.ResponseRecorder, deployed, recorded <-chan struct{}, want int) {
t.Helper()
if w.Code != want {
t.Fatalf("status = %d, want %d (body %q)", w.Code, want, w.Body.String())
}
for what, ch := range map[string]<-chan struct{}{"deploy": deployed, "history write": recorded} {
select {
case <-ch:
t.Errorf("a refused request still did the %s", what)
default:
}
}
release, ok := srv.locks.TryAcquire()
if !ok {
t.Error("a refused request kept a deploy slot")
return
}
release()
}
func fillSlots(t *testing.T, r *reposync.Registry) []func() {
t.Helper()
var held []func()
for len(held) < 64 {
release, ok := r.TryAcquire()
if !ok {
return held
}
held = append(held, release)
}
t.Fatal("the deploy pool never filled")
return nil
}
func TestInternalDeploy_RequiresRepoDid(t *testing.T) {
srv := newTestServer(serverHooks{})
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{}`))
if w.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want 400", w.Code)
}
}
func TestInternalDeploy_AcceptsBeforeTheDeployRuns(t *testing.T) {
var repo models.Repo
var recordCall *worker.DeployContext
var recordArgs [4]string
var ctxCalls, recordCalls int
started := make(chan struct{})
release := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(_ context.Context, repoDid string) (*worker.DeployContext, error) {
ctxCalls++
if repoDid != testRepoDid {
t.Errorf("getCtx repoDid = %q, want %q", repoDid, testRepoDid)
}
return ctxFor(), nil
},
deploy: func(_ context.Context, _ *cloudflare.Client, _ *config.Config, r *models.Repo, branch, dir string) error {
repo = *r
if branch != "main" || dir != "/" {
t.Errorf("deploy branch/dir = %q/%q, want main//", branch, dir)
}
close(started)
<-release
return nil
},
record: func(_ context.Context, dc *worker.DeployContext, sha, status, trigger, errMsg string) error {
recordCalls++
recordCall = dc
recordArgs = [4]string{sha, status, trigger, errMsg}
close(recorded)
return nil
},
})
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
requireAccepted(t, w)
select {
case <-recorded:
t.Fatal("the ack waited for the deploy to finish")
default:
}
await(t, started, "the accepted deploy to run")
close(release)
await(t, recorded, "the history write")
if ctxCalls != 1 {
t.Errorf("getDeployContext calls = %d, want the ack's one preflight", ctxCalls)
}
if recordCalls != 1 {
t.Errorf("recordDeploy calls = %d, want 1", recordCalls)
}
if repo.Did != testOwnerDid || repo.Rkey != "abc1" || repo.Knot != testKnot || repo.RepoDid != testRepoDid {
t.Errorf("deploy repo = %+v, want the owner's rkey/knot/repoDid", repo)
}
if recordCall == nil || recordCall.RepoDid != testRepoDid {
t.Fatalf("record was not called with the repo: %+v", recordCall)
}
if recordArgs[0] != "" || recordArgs[1] != "success" || recordArgs[2] != "config_change" || recordArgs[3] != "" {
t.Errorf("record args = %v, want empty sha, success, config_change, empty error", recordArgs)
}
}
func TestInternalDeploy_Trigger(t *testing.T) {
cases := []struct {
name string
body string
want string
}{
{"omitted is config change", `{"repoDid":"did:plc:repo1"}`, "config_change"},
{"manual", `{"repoDid":"did:plc:repo1","trigger":"manual"}`, "manual"},
{"push", `{"repoDid":"did:plc:repo1","trigger":"push"}`, "push"},
{"unknown is refused", `{"repoDid":"did:plc:repo1","trigger":"cron"}`, ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
var got string
deployed := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
close(deployed)
return nil
},
record: func(_ context.Context, _ *worker.DeployContext, _, _, trigger, _ string) error {
got = trigger
close(recorded)
return nil
},
})
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(tc.body))
if tc.want == "" {
requireRefused(t, srv, w, deployed, recorded, http.StatusBadRequest)
return
}
requireAccepted(t, w)
await(t, recorded, "the history write")
if got != tc.want {
t.Errorf("recorded trigger = %q, want %q", got, tc.want)
}
})
}
}
func TestInternalDeploy_FailureIsRecordedNotReturned(t *testing.T) {
var recordArgs [4]string
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(_ context.Context, _ string) (*worker.DeployContext, error) {
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
return errBoom
},
record: func(_ context.Context, _ *worker.DeployContext, sha, status, trigger, errMsg string) error {
recordArgs = [4]string{sha, status, trigger, errMsg}
close(recorded)
return nil
},
})
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
requireAccepted(t, w)
await(t, recorded, "the history write")
if recordArgs[1] != "failure" || recordArgs[3] != "boom" {
t.Errorf("record args = %v, want failure/boom", recordArgs)
}
}
func TestInternalDeploy_AcksWhileTheDeployIsBlocked(t *testing.T) {
started := make(chan struct{})
release := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
close(started)
<-release
return nil
},
record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
close(recorded)
return nil
},
})
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
requireAccepted(t, w)
select {
case <-recorded:
t.Fatal("the ack carried the deploy's outcome")
default:
}
await(t, started, "the accepted deploy to run")
close(release)
await(t, recorded, "the history write")
}
// negative control: synchronous handler blocks until deploy returns
func TestInternalDeploy_SynchronousShapeStillBlocks(t *testing.T) {
release := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
<-release
return nil
},
record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
close(recorded)
return nil
},
})
w := httptest.NewRecorder()
served := make(chan struct{})
go func() {
defer close(served)
synchronousDeployHandler(srv)(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
}()
awaitQuiet(t, served, 100*time.Millisecond, "the synchronous handler answered while its deploy was blocked")
close(release)
await(t, served, "the synchronous handler's answer")
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want the pre-fix 200 (body %q)", w.Code, w.Body.String())
}
}
func synchronousDeployHandler(s *server) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req deployRequest
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&req); err != nil || req.RepoDid == "" {
writeErr(w, http.StatusBadRequest, "InvalidRequest", "request body must be JSON with a repoDid")
return
}
trigger := models.SiteDeployTriggerConfigChange
if req.Trigger != "" {
var ok bool
if trigger, ok = models.ParseSiteDeployTrigger(req.Trigger); !ok {
writeErr(w, http.StatusBadRequest, "InvalidRequest", "unknown trigger")
return
}
}
done, ok := s.locks.Begin()
if !ok {
writeErr(w, http.StatusServiceUnavailable, "ShuttingDown", "deploy executor is shutting down")
return
}
release, err := s.locks.Acquire(r.Context())
if err != nil {
done()
writeErr(w, http.StatusServiceUnavailable, "DeployUnavailable", err.Error())
return
}
dc, err := s.getCtx(r.Context(), req.RepoDid)
if err != nil || dc == nil {
done()
release()
writeErr(w, http.StatusBadGateway, "DeployFailed", "no deploy context")
return
}
s.runDeploy(acceptedDeploy{dc: dc, trigger: trigger, release: release, done: done})
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(map[string]string{"status": "success"})
}
}
func TestInternalDeploy_DrainWaitsForAcceptedDeploy(t *testing.T) {
started := make(chan struct{})
release := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
close(started)
<-release
return nil
},
record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
close(recorded)
return nil
},
})
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
requireAccepted(t, w)
await(t, started, "the accepted deploy to run")
drained := make(chan struct{})
go func() {
defer close(drained)
srv.locks.Drain()
}()
awaitQuiet(t, drained, 50*time.Millisecond, "Drain returned while an accepted deploy was still running")
close(release)
await(t, drained, "Drain after the accepted deploy")
select {
case <-recorded:
default:
t.Error("Drain returned before the accepted deploy recorded its outcome")
}
}
func TestInternalDeploy_DrainRefusesNewWork(t *testing.T) {
deployed := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
close(deployed)
return nil
},
record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
close(recorded)
return nil
},
})
srv.locks.Drain()
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
requireRefused(t, srv, w, deployed, recorded, http.StatusServiceUnavailable)
}
func TestInternalDeploy_DrainDuringPreflightRefuses(t *testing.T) {
preflight := make(chan struct{})
finishPreflight := make(chan struct{})
deployed := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
close(preflight)
<-finishPreflight
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
close(deployed)
return nil
},
record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
close(recorded)
return nil
},
})
w := httptest.NewRecorder()
served := make(chan struct{})
go func() {
defer close(served)
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
}()
await(t, preflight, "the preflight")
srv.locks.Drain()
close(finishPreflight)
await(t, served, "the refused answer")
requireRefused(t, srv, w, deployed, recorded, http.StatusServiceUnavailable)
}
func TestInternalDeploy_NoDeployContext(t *testing.T) {
cases := []struct {
name string
ctx func() (*worker.DeployContext, error)
want int
}{
{"no site config or active claim", func() (*worker.DeployContext, error) { return nil, nil }, http.StatusUnprocessableEntity},
{"the context lookup failed", func() (*worker.DeployContext, error) { return nil, errBoom }, http.StatusBadGateway},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
deployed := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) { return tc.ctx() },
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
close(deployed)
return nil
},
record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
close(recorded)
return nil
},
})
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
requireRefused(t, srv, w, deployed, recorded, tc.want)
})
}
}
func TestInternalDeploy_RefusesWhenSlotsFull(t *testing.T) {
deployed := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
close(deployed)
return nil
},
record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
close(recorded)
return nil
},
})
held := fillSlots(t, srv.locks)
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
if w.Code != http.StatusTooManyRequests {
t.Fatalf("status = %d, want 429 (body %q)", w.Code, w.Body.String())
}
select {
case <-deployed:
t.Error("a deploy ran without a slot")
default:
}
select {
case <-recorded:
t.Error("a deploy without a slot still wrote history")
default:
}
held[0]()
w = httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
requireAccepted(t, w)
await(t, recorded, "the history write")
}
func TestInternalDeploy_CloudflareNotConfigured(t *testing.T) {
var ctxCalls int
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
ctxCalls++
return ctxFor(), nil
},
})
cf, err := cloudflare.New(&config.Config{})
if err != nil {
t.Fatal(err)
}
if cf.Enabled() {
t.Fatal("an empty cloudflare config must not read as enabled")
}
srv.cf = cf
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
if w.Code != http.StatusServiceUnavailable {
t.Fatalf("status = %d, want 503 (body %q)", w.Code, w.Body.String())
}
if ctxCalls != 0 {
t.Errorf("getDeployContext calls = %d, want none before the cloudflare check", ctxCalls)
}
}
func TestInternalDeploy_RecordsAnAbandonedRepoLock(t *testing.T) {
var recordArgs [4]string
deployed := make(chan struct{})
recorded := make(chan struct{})
srv := newTestServer(serverHooks{
getCtx: func(context.Context, string) (*worker.DeployContext, error) {
return ctxFor(), nil
},
deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
close(deployed)
return nil
},
record: func(_ context.Context, _ *worker.DeployContext, sha, status, trigger, errMsg string) error {
recordArgs = [4]string{sha, status, trigger, errMsg}
close(recorded)
return nil
},
})
unlock, err := srv.locks.Lock(context.Background(), testRepoDid)
if err != nil {
t.Fatal(err)
}
defer unlock()
deployCtx, cancel := context.WithCancel(context.Background())
cancel()
srv.deployCtx = deployCtx
w := httptest.NewRecorder()
srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
requireAccepted(t, w)
await(t, recorded, "the abandoned deploy's history write")
if recordArgs[1] != "failure" || !strings.Contains(recordArgs[3], "repo lock") {
t.Errorf("record args = %v, want a failure naming the abandoned repo lock", recordArgs)
}
select {
case <-deployed:
t.Error("the deploy ran without the repo lock")
default:
}
}