// provision_test.go proves RFC 0011 §7/Phase 0 end to end: quarry, not
// an operator, keeps notify's subscribers, topic membership and
// workflows in step with its own accounts and grants, and a chat post's
// mentions reach only people notify can actually address. The fake
// notify server here is built against notify's documented routes
// (docs/running.md, pre-0011-phase-0 revision) — regenerate it from
// notify's OpenAPI if those routes change (RFC 0012 §Risks).
package harness
import (
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"net/url"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/chat"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/notify"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/web"
"craftdesign.group/quarry/internal/zitcli"
)
// fakeNotifyReq is one call the fake notify server saw.
type fakeNotifyReq struct {
method, path string
query string
body map[string]any
}
// fakeNotifyServer is a recording stand-in for notify's HTTP surface:
// every request is remembered, and workflows are stateful (a slug can
// be pre-seeded with an older definition, the way a forge provisioned
// by an earlier quarry holds one) so EnsureWorkflow's in-place upsert
// is exercised for real. Subscriber existence is stateful too, and topic
// membership enforces it: notify's real POST
// /v1/topics/{key}/subscribers answers 422 "subscriber not found" for
// a subscriberId nothing ever created (httpapi.go's postTopicSubscriber)
// — modeled here so a test can prove that a grant on an account that
// was never provisioned actually fails the way it does against real
// notify, instead of the fake silently succeeding at everything.
type fakeNotifyServer struct {
mu sync.Mutex
reqs []fakeNotifyReq
workflows map[string]*fakeWorkflow
subscribers map[string]bool
// updated names the slugs an upsert rewrote in place; deletes counts
// DELETEs, which would lose the preferences notify keys by id.
updated []string
deletes int
}
// fakeWorkflow is a stored workflow: an id that never changes, and the
// name+steps notify's upsert compares.
type fakeWorkflow struct {
id, def string
}
func fakeWorkflowDef(body map[string]any) string {
b, _ := json.Marshal(map[string]any{"name": body["name"], "steps": body["steps"]})
return string(b)
}
func (f *fakeNotifyServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
var body map[string]any
if r.Body != nil {
b, _ := io.ReadAll(r.Body)
if len(b) > 0 {
json.Unmarshal(b, &body)
}
}
f.mu.Lock()
f.reqs = append(f.reqs, fakeNotifyReq{method: r.Method, path: r.URL.Path,
query: r.URL.RawQuery, body: body})
defer f.mu.Unlock()
switch {
case r.Method == "GET" && strings.HasPrefix(r.URL.Path, "/v1/workflows/"):
slug := strings.TrimPrefix(r.URL.Path, "/v1/workflows/")
if f.workflows[slug] != nil {
w.WriteHeader(http.StatusOK)
return
}
w.WriteHeader(http.StatusNotFound)
case r.Method == "DELETE" && strings.HasPrefix(r.URL.Path, "/v1/workflows/"):
f.deletes++
w.WriteHeader(http.StatusNoContent)
case r.Method == "PUT" && strings.HasPrefix(r.URL.Path, "/v1/workflows/"):
// notify#303's upsert: create when absent, no write when the
// definition matches, and when it differs `update` rewrites the
// row in place (same id) while `keep` answers 409.
slug := strings.TrimPrefix(r.URL.Path, "/v1/workflows/")
def := fakeWorkflowDef(body)
reply := func(status int, outcome string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
json.NewEncoder(w).Encode(map[string]any{"_id": f.workflows[slug].id, "outcome": outcome})
}
switch wf := f.workflows[slug]; {
case wf == nil:
f.workflows[slug] = &fakeWorkflow{id: "wf-" + slug, def: def}
reply(http.StatusCreated, "created")
case wf.def == def:
reply(http.StatusOK, "unchanged")
case r.URL.Query().Get("onConflict") == "keep":
w.WriteHeader(http.StatusConflict)
default:
wf.def = def
f.updated = append(f.updated, slug)
reply(http.StatusOK, "updated")
}
case r.Method == "POST" && r.URL.Path == "/v1/workflows":
name, _ := body["name"].(string)
f.workflows[name] = &fakeWorkflow{id: "wf-" + name, def: fakeWorkflowDef(body)}
w.WriteHeader(http.StatusCreated)
case r.Method == "POST" && r.URL.Path == "/v1/subscribers":
id, _ := body["subscriberId"].(string)
if id != "" {
if f.subscribers == nil {
f.subscribers = map[string]bool{}
}
f.subscribers[id] = true
}
w.WriteHeader(http.StatusCreated)
case r.Method == "POST" && strings.HasSuffix(r.URL.Path, "/subscribers") && strings.HasPrefix(r.URL.Path, "/v1/topics/"):
id, _ := body["subscriberId"].(string)
if !f.subscribers[id] {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusUnprocessableEntity)
json.NewEncoder(w).Encode(map[string]any{
"error": "unprocessable_entity", "message": "subscriber not found",
})
return
}
w.WriteHeader(http.StatusCreated)
default:
// topics and events/trigger just need to be recorded and
// answered success.
w.WriteHeader(http.StatusCreated)
}
}
func (f *fakeNotifyServer) snapshot() []fakeNotifyReq {
f.mu.Lock()
defer f.mu.Unlock()
return append([]fakeNotifyReq(nil), f.reqs...)
}
func (f *fakeNotifyServer) find(method, path string) []fakeNotifyReq {
var out []fakeNotifyReq
for _, r := range f.snapshot() {
if r.method == method && r.path == path {
out = append(out, r)
}
}
return out
}
// waitForCount polls until n's request count for (method, path) reaches
// want or the deadline passes — the same shape notifyCapture.waitFor
// uses, because provisioning here is exactly as best-effort-async as a
// notify trigger.
func waitForCount(t *testing.T, srv *fakeNotifyRecorder, method, path string, want int) {
t.Helper()
for i := 0; i < 100; i++ {
if len(srv.f.find(method, path)) >= want {
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("never saw %d %s %s (saw %+v)", want, method, path, srv.f.snapshot())
}
// fakeNotifyRecorder pairs the httptest server with its handler so
// tests can both hit it over HTTP (via notify.Client) and inspect what
// it recorded.
type fakeNotifyRecorder struct {
srv *httptest.Server
f *fakeNotifyServer
}
func startFakeNotify(t *testing.T, preseed ...string) *fakeNotifyRecorder {
t.Helper()
f := &fakeNotifyServer{workflows: map[string]*fakeWorkflow{}}
for _, w := range preseed {
f.workflows[w] = &fakeWorkflow{id: "wf-" + w, def: `{"name":"an older definition"}`}
}
srv := httptest.NewServer(f)
t.Cleanup(srv.Close)
return &fakeNotifyRecorder{srv: srv, f: f}
}
// harnessAccounts adapts forge OLTP to api.AccountStore, mirroring
// cmd/quarry's own accountStore (unexported there, so restated here —
// this IS the seam a fresh harness test needs, per the brief).
type harnessAccounts struct{ db *oltp.DB }
func (a harnessAccounts) Accounts() ([]api.Account, error) {
rows, err := a.db.Accounts()
if err != nil {
return nil, err
}
out := make([]api.Account, 0, len(rows))
for _, r := range rows {
out = append(out, api.Account{Name: r.Name, Admin: r.Admin, Created: r.Created})
}
return out, nil
}
func (a harnessAccounts) AddAccount(name, password string, admin bool) error {
return a.db.AddAccount(name, password, admin)
}
func (a harnessAccounts) SetPassword(name, password string) error {
return a.db.SetPassword(name, password)
}
func (a harnessAccounts) SetOperator(name, operator string) error {
return a.db.SetOperator(name, operator)
}
func (a harnessAccounts) DeleteAccount(name string) error { return a.db.DeleteAccount(name) }
func (a harnessAccounts) Grant(actor, account, repo, role string) (string, error) {
r, err := oltp.ParseRole(role)
if err != nil {
return "", err
}
return a.db.GrantGuarded(actor, account, repo, r)
}
func (a harnessAccounts) Revoke(actor, account, repo string) (string, error) {
return a.db.GrantGuarded(actor, account, repo, oltp.NoRole)
}
func (a harnessAccounts) GrantsFor(account string) (map[string]string, error) {
grants, err := a.db.GrantsFor(account)
if err != nil {
return nil, err
}
out := make(map[string]string, len(grants))
for repo, role := range grants {
out[repo] = role.String()
}
return out, nil
}
// harnessProvision adapts api.ProvisionPlane to the notify client,
// mirroring cmd/quarry's provisionAdapter.
type harnessProvision struct{ client *notify.Client }
func (p *harnessProvision) EnsureSubscriber(ctx context.Context, account string) error {
return p.client.EnsureSubscriber(ctx, account, account+"@example.test")
}
func (p *harnessProvision) RemoveSubscriber(ctx context.Context, account string) error {
return p.client.RemoveSubscriber(ctx, account)
}
func (p *harnessProvision) SetTopicMember(ctx context.Context, repo, account string, member bool) error {
key := "repo-" + repo
if err := p.client.EnsureTopic(ctx, key, repo+" activity"); err != nil {
return err
}
if member {
// Mirrors cmd/quarry's provisionAdapter.SetTopicMember: upsert
// the subscriber before adding it to the topic, since notify
// 422s a topic-member add for a subscriberId nothing created.
if err := p.EnsureSubscriber(ctx, account); err != nil {
return err
}
return p.client.AddTopicMember(ctx, key, account)
}
return p.client.RemoveTopicMember(ctx, key, account)
}
// harnessNotify adapts api.NotifyPlane to the notify client, mirroring
// cmd/quarry's notifyAdapter but without the repo: topic addressing
// (Phase 0's mention triggers always address a person, never a topic).
type harnessNotify struct{ client *notify.Client }
func (n *harnessNotify) Trigger(ctx context.Context, e api.NotifyEvent) error {
return n.client.Trigger(ctx, e.Workflow, notify.Recipient{SubscriberID: e.To}, e.Payload, e.TxnID)
}
// buildProvisionServer stands up a real forge (one repo, real chat and
// accounts planes, real notify.Client wired to the fake server) — the
// harness proof this phase's spec asks for, built independently of
// ingestAndServeAuthMod because wiring Accounts/CapsFor there would
// retroactively filter mentions in every OTHER harness test that
// mentions a handle with no real account row (TestForumSubscriptions'
// @carol, notably).
func buildProvisionServer(t *testing.T, notifyURL string) (*httptest.Server, *oltp.DB) {
t.Helper()
dataDir := t.TempDir()
ev, err := events.Start(filepath.Join(dataDir, "nats"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
odb, err := oltp.Open(filepath.Join(dataDir, "oltp.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { odb.Close() })
room, err := chat.New(ev.Conn(), ev.JS())
if err != nil {
t.Fatal(err)
}
nclient := notify.New(notifyURL, "k")
deps := api.Deps{
Version: "0.0.0-harness",
Repos: api.NewRepos(nil),
Chat: room,
Notify: &harnessNotify{client: nclient},
Provision: &harnessProvision{client: nclient},
Accounts: harnessAccounts{db: odb},
Forum: odb,
}
deps.CapsFor = func(account, repo string) api.CapSet { return odb.EffectiveCaps(account, repo) }
repoPath := buildStore(t, "provseed")
ix, err := index.Open(filepath.Join(dataDir, "alpha.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
cli := zitcli.New(repoPath)
pipe := ingest.NewPipeline(cli, ev, ix, "alpha", "harness")
deps.Repos.Put("alpha", &api.Repo{Index: ix, Zit: cli, Sync: pipe.Sync, Code: "alpha"})
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
// dan bootstraps as the first admin exactly as docs/running.md's
// `quarry account add --admin` does — an existing admin account is
// the precondition for account.add/grant/revoke to work at all.
if err := odb.AddAccount("dan", "pw", true); err != nil {
t.Fatal(err)
}
// Granting on alpha is alpha's business, and a site admin holds
// nothing there unless granted (#318): dan administers it too.
if err := odb.Grant("dan", "alpha", oltp.RepoAdmin); err != nil {
t.Fatal(err)
}
verify := func(repo, user, pass string, need ops.Auth) api.Decision {
if user == "" || pass != "pw" {
return api.Denied
}
return api.Allowed
}
r := ops.NewRegistry()
if err := api.RegisterAll(r, deps); err != nil {
t.Fatal(err)
}
apiRouter, err := api.RouterWithCaps(r, "0.0.0-harness", deps.CapsFor, deps.CanRead, verify)
if err != nil {
t.Fatal(err)
}
mux := http.NewServeMux()
mux.Handle("/api/", apiRouter)
mux.Handle("/", web.HandlerWith(r, nil, nil, nil, web.Verify(verify)))
srv := httptest.NewServer(mux)
t.Cleanup(srv.Close)
return srv, odb
}
func TestProvisionEndToEnd(t *testing.T) {
fake := startFakeNotify(t, "quarry-mention") // pre-seed one workflow as "already there"
srv, _ := buildProvisionServer(t, fake.srv.URL)
client := srv.Client()
call := func(user, method, path string, form url.Values) (int, []byte) {
t.Helper()
var body io.Reader
if form != nil {
body = strings.NewReader(form.Encode())
}
req, _ := http.NewRequest(method, srv.URL+path, body)
if user != "" {
req.SetBasicAuth(user, "pw")
}
if form != nil {
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
}
resp, err := client.Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
b, _ := io.ReadAll(resp.Body)
return resp.StatusCode, b
}
// --- startup: EnsureWorkflow for each of notify.Workflows(), as
// cmd/quarry's serve does right after the notify client exists.
nclient := notify.New(fake.srv.URL, "k")
staleID := fake.f.workflows["quarry-mention"].id
boot := func() {
t.Helper()
for _, wf := range notify.Workflows() {
if _, err := nclient.EnsureWorkflow(context.Background(), wf); err != nil {
t.Fatalf("EnsureWorkflow(%s): %v", wf.Name, err)
}
}
}
boot()
// Every workflow was upserted with `update`: the missing ones were
// created, and the pre-seeded one — an older definition — was
// rewritten in place, keeping its id.
upserted := map[string]bool{}
for _, r := range fake.f.snapshot() {
if r.method != "PUT" || !strings.HasPrefix(r.path, "/v1/workflows/") {
continue
}
if r.query != "onConflict=update" {
t.Fatalf("a workflow upsert carried %q; want onConflict=update", r.query)
}
upserted[strings.TrimPrefix(r.path, "/v1/workflows/")] = true
}
if got := fake.f.updated; len(got) != 1 || got[0] != "quarry-mention" {
t.Fatalf("updated = %v, want only the stale quarry-mention", got)
}
if got := fake.f.workflows["quarry-mention"].id; got != staleID {
t.Fatalf("quarry-mention id %q -> %q; preferences are keyed by it", staleID, got)
}
for _, wf := range notify.Workflows() {
if !upserted[wf.TriggerIdentifier] {
t.Fatalf("workflow %q was never upserted", wf.TriggerIdentifier)
}
}
// A second start finds every definition current and rewrites none.
boot()
if got := fake.f.updated; len(got) != 1 {
t.Fatalf("a second start rewrote workflows: %v", got)
}
if fake.f.deletes != 0 {
t.Fatalf("provisioning sent %d workflow DELETEs", fake.f.deletes)
}
// And nothing went through the old create route.
if got := fake.f.find("POST", "/v1/workflows"); len(got) > 0 {
t.Fatalf("provisioning still POSTs workflows: %+v", got)
}
// --- account.add creates the subscriber.
if code, body := call("dan", "POST", "/api/v1/accounts",
url.Values{"name": {"alice"}, "password": {"pw12345678"}}); code != http.StatusOK {
t.Fatalf("account.add: %d: %s", code, body)
}
waitForCount(t, fake, "POST", "/v1/subscribers", 1)
found := false
for _, r := range fake.f.find("POST", "/v1/subscribers") {
if r.body["subscriberId"] == "alice" {
found = true
}
}
if !found {
t.Fatalf("no subscriber request for alice: %+v", fake.f.find("POST", "/v1/subscribers"))
}
// --- account.grant adds alice to repo-alpha's topic.
if code, body := call("dan", "POST", "/api/v1/accounts/grant",
url.Values{"name": {"alice"}, "repo": {"alpha"}, "role": {"reader"}}); code != http.StatusOK {
t.Fatalf("account.grant: %d: %s", code, body)
}
waitForCount(t, fake, "POST", "/v1/topics/repo-alpha/subscribers", 1)
found = false
for _, r := range fake.f.find("POST", "/v1/topics/repo-alpha/subscribers") {
if r.body["subscriberId"] == "alice" {
found = true
}
}
if !found {
t.Fatal("alice was never added to repo-alpha's topic")
}
// --- account.revoke removes alice from the topic.
if code, body := call("dan", "POST", "/api/v1/accounts/revoke",
url.Values{"name": {"alice"}, "repo": {"alpha"}}); code != http.StatusOK {
t.Fatalf("account.revoke: %d: %s", code, body)
}
waitForCount(t, fake, "DELETE", "/v1/topics/repo-alpha/subscribers/alice", 1)
// alice needs to be re-granted for the mention tests below (revoke
// above proved removal; the mention path needs her capable again).
if code, body := call("dan", "POST", "/api/v1/accounts/grant",
url.Values{"name": {"alice"}, "repo": {"alpha"}, "role": {"reader"}}); code != http.StatusOK {
t.Fatalf("account.grant (re-grant): %d: %s", code, body)
}
// --- anonymous chat.post is refused.
if code, _ := call("", "POST", "/api/v1/r/alpha/chat", url.Values{"text": {"hello"}}); code != http.StatusUnauthorized && code != http.StatusForbidden {
t.Fatalf("anonymous chat.post: status %d, want 401 or 403", code)
}
// --- @unknown triggers nothing.
before := len(fake.f.find("POST", "/v1/events/trigger"))
if code, body := call("dan", "POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"hi @nobody"}}); code != http.StatusOK {
t.Fatalf("chat.post: %d: %s", code, body)
}
time.Sleep(200 * time.Millisecond) // fireNotify is async; give it a beat
if got := len(fake.f.find("POST", "/v1/events/trigger")); got != before {
t.Fatalf("@nobody fired %d triggers, want 0 new", got-before)
}
// --- @alice (granted, capable) triggers exactly once with an
// address-based TxnID.
if code, body := call("dan", "POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"hey @alice"}}); code != http.StatusOK {
t.Fatalf("chat.post: %d: %s", code, body)
}
waitForCount(t, fake, "POST", "/v1/events/trigger", before+1)
var toAlice []fakeNotifyReq
for _, r := range fake.f.find("POST", "/v1/events/trigger")[before:] {
to, _ := r.body["to"].(map[string]any)
if to != nil && to["subscriberId"] == "alice" {
toAlice = append(toAlice, r)
}
}
if len(toAlice) != 1 {
t.Fatalf("expected exactly one trigger to alice, got %d: %+v", len(toAlice), toAlice)
}
txn, _ := toAlice[0].body["transactionId"].(string)
if !strings.HasPrefix(txn, "mention:chat:alpha:") || !strings.HasSuffix(txn, ":alice") {
t.Fatalf("TxnID %q is not address-based (mention:chat:<repo>:<at>:<who>)", txn)
}
// --- account.delete removes the subscriber, so a deleted person's
// address does not stay in notify.
if code, body := call("dan", "POST", "/api/v1/accounts/delete",
url.Values{"name": {"alice"}}); code != http.StatusOK {
t.Fatalf("account.delete: %d: %s", code, body)
}
waitForCount(t, fake, "DELETE", "/v1/subscribers/alice", 1)
}
// TestProvisionGrantEnsuresPreexistingSubscriber proves the fix for the
// gap account.add alone leaves open: an account that exists in OLTP but
// was never run through account.add — imported from fossil, restored
// from a backup, or simply present before this feature shipped — has no
// notify subscriber. Against real notify, granting such an account
// answers 422 "subscriber not found" on the topic-member add (modeled
// by fakeNotifyServer above) and membership never happens. SetTopicMember
// now upserts the subscriber on its own add path, so the grant must
// succeed anyway.
func TestProvisionGrantEnsuresPreexistingSubscriber(t *testing.T) {
fake := startFakeNotify(t, notify.Workflows()[0].TriggerIdentifier)
srv, odb := buildProvisionServer(t, fake.srv.URL)
client := srv.Client()
call := func(user, method, path string, form url.Values) (int, []byte) {
t.Helper()
var body io.Reader
if form != nil {
body = strings.NewReader(form.Encode())
}
req, _ := http.NewRequest(method, srv.URL+path, body)
if user != "" {
req.SetBasicAuth(user, "pw")
}
if form != nil {
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
}
resp, err := client.Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
b, _ := io.ReadAll(resp.Body)
return resp.StatusCode, b
}
// bob exists in OLTP but never went through account.add — notify has
// never heard of him. This is the "pre-existing account" case: an
// import, a restore, or an account older than RFC 0011 §7.
if err := odb.AddAccount("bob", "pw12345678", false); err != nil {
t.Fatal(err)
}
if len(fake.f.find("POST", "/v1/subscribers")) != 0 {
t.Fatal("bob must not already be a notify subscriber for this test to prove anything")
}
if code, body := call("dan", "POST", "/api/v1/accounts/grant",
url.Values{"name": {"bob"}, "repo": {"alpha"}, "role": {"reader"}}); code != http.StatusOK {
t.Fatalf("account.grant: %d: %s", code, body)
}
waitForCount(t, fake, "POST", "/v1/topics/repo-alpha/subscribers", 1)
subscribed := false
for _, r := range fake.f.find("POST", "/v1/subscribers") {
if r.body["subscriberId"] == "bob" {
subscribed = true
}
}
if !subscribed {
t.Fatalf("bob was never provisioned as a subscriber before the topic-member add: %+v",
fake.f.find("POST", "/v1/subscribers"))
}
member := false
for _, r := range fake.f.find("POST", "/v1/topics/repo-alpha/subscribers") {
if r.body["subscriberId"] == "bob" {
member = true
}
}
if !member {
t.Fatal("bob was never added to repo-alpha's topic")
}
}