package notify
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
)
// recorder is a minimal fake notify server: it remembers every request
// it saw and answers with whatever statuses the test primes.
type recorder struct {
reqs []recordedReq
statuses map[string]int // method+" "+path -> status to answer once, then 200
}
type recordedReq struct {
method, path string
// query is the raw query string, which is where notify#303's
// onConflict mode lives — the difference between "create if absent"
// and "overwrite whatever the operator edited".
query string
body map[string]any
}
func (r *recorder) ServeHTTP(w http.ResponseWriter, req *http.Request) {
var body map[string]any
json.NewDecoder(req.Body).Decode(&body)
r.reqs = append(r.reqs, recordedReq{method: req.Method, path: req.URL.Path,
query: req.URL.RawQuery, body: body})
key := req.Method + " " + req.URL.Path
if status, ok := r.statuses[key]; ok {
w.WriteHeader(status)
return
}
w.WriteHeader(http.StatusOK)
}
func serverWith(t *testing.T, handler http.HandlerFunc) (*httptest.Server, *recorder) {
t.Helper()
rec := &recorder{statuses: map[string]int{}}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
rec.ServeHTTP(w, req)
if handler != nil {
handler(w, req)
}
}))
t.Cleanup(srv.Close)
return srv, rec
}
func TestEnsureSubscriber(t *testing.T) {
srv, rec := serverWith(t, nil)
c := New(srv.URL, "k")
if err := c.EnsureSubscriber(context.Background(), "alice", ""); err != nil {
t.Fatal(err)
}
if len(rec.reqs) != 1 || rec.reqs[0].path != "/v1/subscribers" {
t.Fatalf("requests = %+v", rec.reqs)
}
if rec.reqs[0].body["subscriberId"] != "alice" {
t.Fatalf("body = %+v", rec.reqs[0].body)
}
}
func TestEnsureSubscriberNilClient(t *testing.T) {
var c *Client
if err := c.EnsureSubscriber(context.Background(), "alice", ""); err != nil {
t.Fatalf("nil client must no-op: %v", err)
}
c = New("", "k")
if err := c.EnsureSubscriber(context.Background(), "alice", ""); err != nil {
t.Fatalf("disabled client must no-op: %v", err)
}
}
func TestEnsureSubscriberAlreadyExists(t *testing.T) {
rec := &recorder{statuses: map[string]int{"POST /v1/subscribers": http.StatusConflict}}
srv := httptest.NewServer(rec)
t.Cleanup(srv.Close)
c := New(srv.URL, "k")
if err := c.EnsureSubscriber(context.Background(), "alice", ""); err != nil {
t.Fatalf("409 must be treated as success: %v", err)
}
}
func TestEnsureSubscriberRealFailure(t *testing.T) {
rec := &recorder{statuses: map[string]int{"POST /v1/subscribers": http.StatusInternalServerError}}
srv := httptest.NewServer(rec)
t.Cleanup(srv.Close)
c := New(srv.URL, "k")
if err := c.EnsureSubscriber(context.Background(), "alice", ""); err == nil {
t.Fatal("a real 500 must be reported")
}
}
func TestEnsureTopic(t *testing.T) {
srv, rec := serverWith(t, nil)
c := New(srv.URL, "k")
if err := c.EnsureTopic(context.Background(), "repo-quarry", "quarry activity"); err != nil {
t.Fatal(err)
}
if len(rec.reqs) != 1 || rec.reqs[0].path != "/v1/topics" ||
rec.reqs[0].body["key"] != "repo-quarry" || rec.reqs[0].body["name"] != "quarry activity" {
t.Fatalf("requests = %+v", rec.reqs)
}
}
func TestAddTopicMember(t *testing.T) {
srv, rec := serverWith(t, nil)
c := New(srv.URL, "k")
if err := c.AddTopicMember(context.Background(), "repo-quarry", "alice"); err != nil {
t.Fatal(err)
}
if len(rec.reqs) != 1 || rec.reqs[0].path != "/v1/topics/repo-quarry/subscribers" ||
rec.reqs[0].body["subscriberId"] != "alice" {
t.Fatalf("requests = %+v", rec.reqs)
}
}
func TestAddTopicMemberAlreadyExists(t *testing.T) {
rec := &recorder{statuses: map[string]int{"POST /v1/topics/repo-quarry/subscribers": http.StatusConflict}}
srv := httptest.NewServer(rec)
t.Cleanup(srv.Close)
c := New(srv.URL, "k")
if err := c.AddTopicMember(context.Background(), "repo-quarry", "alice"); err != nil {
t.Fatalf("409 must be treated as success: %v", err)
}
}
func TestRemoveTopicMember(t *testing.T) {
rec := &recorder{statuses: map[string]int{}}
srv := httptest.NewServer(rec)
t.Cleanup(srv.Close)
c := New(srv.URL, "k")
if err := c.RemoveTopicMember(context.Background(), "repo-quarry", "alice"); err != nil {
t.Fatal(err)
}
if len(rec.reqs) != 1 || rec.reqs[0].method != "DELETE" ||
rec.reqs[0].path != "/v1/topics/repo-quarry/subscribers/alice" {
t.Fatalf("requests = %+v", rec.reqs)
}
}
func TestRemoveSubscriber(t *testing.T) {
rec := &recorder{statuses: map[string]int{}}
srv := httptest.NewServer(rec)
t.Cleanup(srv.Close)
c := New(srv.URL, "k")
if err := c.RemoveSubscriber(context.Background(), "alice"); err != nil {
t.Fatal(err)
}
if len(rec.reqs) != 1 || rec.reqs[0].method != "DELETE" || rec.reqs[0].path != "/v1/subscribers/alice" {
t.Fatalf("requests = %+v", rec.reqs)
}
rec.statuses["DELETE /v1/subscribers/bob"] = http.StatusNotFound
if err := c.RemoveSubscriber(context.Background(), "bob"); err != nil {
t.Fatalf("404 on remove must be treated as success: %v", err)
}
}
func TestRemoveTopicMemberAbsentIsSuccess(t *testing.T) {
rec := &recorder{statuses: map[string]int{"DELETE /v1/topics/repo-quarry/subscribers/alice": http.StatusNotFound}}
srv := httptest.NewServer(rec)
t.Cleanup(srv.Close)
c := New(srv.URL, "k")
if err := c.RemoveTopicMember(context.Background(), "repo-quarry", "alice"); err != nil {
t.Fatalf("404 on remove must be treated as success: %v", err)
}
}
// workflowStore models notify's PUT /v1/workflows/{slug} upsert
// (authoring.Ensure): a row keeps its id for life, an identical
// definition is not written, a different one is rewritten in place
// under onConflict=update. A DELETE is recorded and refused — it is
// the one call that would lose subscribers' per-workflow preferences,
// which notify keys by the workflow's id.
type workflowStore struct {
mu sync.Mutex
rows map[string]*storedWorkflow
reqs []recordedReq
deletes int
nextID int
}
type storedWorkflow struct {
id string
def string // name+steps as JSON, the part Ensure compares
version int
}
func defJSON(name string, steps any) string {
// Round-trip through any so typed and decoded steps compare alike.
var canon any
b, _ := json.Marshal(map[string]any{"name": name, "steps": steps})
json.Unmarshal(b, &canon)
b, _ = json.Marshal(canon)
return string(b)
}
func (s *workflowStore) seed(def WorkflowDef) {
s.mu.Lock()
defer s.mu.Unlock()
s.nextID++
s.rows[def.TriggerIdentifier] = &storedWorkflow{id: fmt.Sprintf("wf-%d", s.nextID),
def: defJSON(def.Name, def.Steps), version: 1}
}
func (s *workflowStore) row(slug string) storedWorkflow {
s.mu.Lock()
defer s.mu.Unlock()
return *s.rows[slug]
}
func (s *workflowStore) ServeHTTP(w http.ResponseWriter, r *http.Request) {
var body map[string]any
json.NewDecoder(r.Body).Decode(&body)
s.mu.Lock()
defer s.mu.Unlock()
s.reqs = append(s.reqs, recordedReq{method: r.Method, path: r.URL.Path,
query: r.URL.RawQuery, body: body})
slug := strings.TrimPrefix(r.URL.Path, "/v1/workflows/")
if r.Method == "DELETE" {
s.deletes++
w.WriteHeader(http.StatusForbidden)
return
}
if r.Method != "PUT" {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
name, _ := body["name"].(string)
def := defJSON(name, body["steps"])
reply := func(status int, row *storedWorkflow, outcome string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
json.NewEncoder(w).Encode(map[string]any{"_id": row.id, "triggerIdentifier": slug, "outcome": outcome})
}
row, ok := s.rows[slug]
switch {
case !ok:
s.nextID++
row = &storedWorkflow{id: fmt.Sprintf("wf-%d", s.nextID), def: def, version: 1}
s.rows[slug] = row
reply(http.StatusCreated, row, "created")
case row.def == def:
reply(http.StatusOK, row, "unchanged")
case r.URL.Query().Get("onConflict") == "keep":
w.WriteHeader(http.StatusConflict)
default:
row.def = def
row.version++
reply(http.StatusOK, row, "updated")
}
}
func startWorkflowStore(t *testing.T) (*workflowStore, *Client) {
t.Helper()
s := &workflowStore{rows: map[string]*storedWorkflow{}}
srv := httptest.NewServer(s)
t.Cleanup(srv.Close)
return s, New(srv.URL, "k")
}
func mentionDef(body string) WorkflowDef {
return WorkflowDef{Name: "quarry-mention", TriggerIdentifier: "quarry-mention",
Steps: []WorkflowStep{{StepID: "email", Type: "email",
ControlValues: map[string]any{"subject": "s", "body": "b", "htmlBody": body}}}}
}
// Provisioning runs on every start: a definition notify already holds
// must come back unchanged, with nothing rewritten.
func TestEnsureWorkflowUnchangedWritesNothing(t *testing.T) {
s, c := startWorkflowStore(t)
s.seed(mentionDef("<p>b</p>"))
before := s.row("quarry-mention")
got, err := c.EnsureWorkflow(context.Background(), mentionDef("<p>b</p>"))
if err != nil {
t.Fatal(err)
}
if got != WorkflowUnchanged {
t.Fatalf("outcome = %q, want %q", got, WorkflowUnchanged)
}
if after := s.row("quarry-mention"); after != before {
t.Fatalf("an identical definition was rewritten: %+v -> %+v", before, after)
}
if len(s.reqs) != 1 {
t.Fatalf("requests = %+v", s.reqs)
}
}
// A template quarry changed reaches notify in place: one PUT with
// onConflict=update, the same workflow id, and never a DELETE.
func TestEnsureWorkflowUpdatesChangedTemplateInPlace(t *testing.T) {
s, c := startWorkflowStore(t)
s.seed(mentionDef("<p>old</p>"))
before := s.row("quarry-mention")
got, err := c.EnsureWorkflow(context.Background(), mentionDef("<p>new</p>"))
if err != nil {
t.Fatal(err)
}
if got != WorkflowUpdated {
t.Fatalf("outcome = %q, want %q", got, WorkflowUpdated)
}
after := s.row("quarry-mention")
if after.id != before.id {
t.Fatalf("workflow id changed %q -> %q; preferences are keyed by it", before.id, after.id)
}
if after.version != before.version+1 || after.def != defJSON("quarry-mention", mentionDef("<p>new</p>").Steps) {
t.Fatalf("stored row = %+v, want the new definition written once", after)
}
if s.deletes != 0 {
t.Fatalf("provisioning sent %d DELETEs", s.deletes)
}
if len(s.reqs) != 1 {
t.Fatalf("requests = %+v", s.reqs)
}
r := s.reqs[0]
if r.method != "PUT" || r.path != "/v1/workflows/quarry-mention" || r.query != "onConflict=update" {
t.Fatalf("request = %+v", r)
}
}
func TestEnsureWorkflowCreatesMissing(t *testing.T) {
s, c := startWorkflowStore(t)
got, err := c.EnsureWorkflow(context.Background(), mentionDef("<p>b</p>"))
if err != nil {
t.Fatal(err)
}
if got != WorkflowCreated {
t.Fatalf("outcome = %q, want %q", got, WorkflowCreated)
}
if s.row("quarry-mention").version != 1 {
t.Fatalf("row = %+v", s.row("quarry-mention"))
}
}
// A notify that is down is an error for the caller to log, returned
// promptly — never a hang or a panic on the startup goroutine.
func TestEnsureWorkflowNotifyDown(t *testing.T) {
srv := httptest.NewServer(http.NotFoundHandler())
url := srv.URL
srv.Close()
c := New(url, "k")
got, err := c.EnsureWorkflow(context.Background(), mentionDef("<p>b</p>"))
if err == nil {
t.Fatal("an unreachable notify must be an error")
}
if got != "" {
t.Fatalf("outcome = %q on failure", got)
}
}
// A notify that predates the outcome field still answers 201 for a
// create and 200 for everything else.
func TestEnsureWorkflowOutcomeFromStatus(t *testing.T) {
for status, want := range map[int]WorkflowOutcome{
http.StatusCreated: WorkflowCreated, http.StatusOK: WorkflowUnchanged} {
rec := &recorder{statuses: map[string]int{"PUT /v1/workflows/quarry-mention": status}}
srv := httptest.NewServer(rec)
c := New(srv.URL, "k")
got, err := c.EnsureWorkflow(context.Background(), mentionDef("<p>b</p>"))
srv.Close()
if err != nil || got != want {
t.Fatalf("status %d: outcome = %q, err = %v; want %q", status, got, err, want)
}
}
}
func TestEnsureWorkflowRefusalIsError(t *testing.T) {
rec := &recorder{statuses: map[string]int{"PUT /v1/workflows/quarry-mention": http.StatusUnprocessableEntity}}
srv := httptest.NewServer(rec)
t.Cleanup(srv.Close)
if _, err := New(srv.URL, "k").EnsureWorkflow(context.Background(), mentionDef("<p>b</p>")); err == nil {
t.Fatal("a 422 must be an error")
}
}
func TestEnsureWorkflowNilClient(t *testing.T) {
var c *Client
if _, err := c.EnsureWorkflow(context.Background(), WorkflowDef{TriggerIdentifier: "x"}); err != nil {
t.Fatalf("nil client must no-op: %v", err)
}
}
// The routing table of RFC 0011 §6, asserted as step TYPES in order —
// the shape notify runs, not the copy, which is an operator's to edit.
func TestWorkflowsCoverTheKnownNames(t *testing.T) {
want := map[string][]string{
// Email LAST and off by default: the step exists so the choice
// exists (emailOffByDefault writes the "no" for everyone who
// has not chosen).
"quarry-mention": {"in_app", "push", "email"},
"quarry-forum-reply": {"in_app", "push", "email"},
"quarry-landing": {"in_app", "push", "email"},
// A digest BEFORE the in_app step: the inbox row is the batch.
"quarry-activity": {"digest", "in_app", "email"},
"quarry-error": {"in_app"},
"quarry-slow": {"in_app"},
// The host's own alarms (a deploy that failed, a forge that
// crashed), sent by scripts/quarry-alert.sh to the admins
// topic: push, because the person has to hear it now.
"quarry-ops": {"in_app", "push"},
// IN-APP ONLY, and the absence of an email step is the point:
// an email step must belong to a preference family or it is
// mail nobody can turn off. The defect this workflow fixes is
// the missing inbox row — an author was never told their held
// submission had been rejected, while the reason for it sat in
// a public log addressed to everybody (b32aeba12bc6).
"quarry-moderation": {"in_app"},
// No quarry-invite: the invitation is sent by quarry's own
// mailer, which holds the credential and the copy, and an
// invitee has no account for any other channel to reach. A
// workflow here as well meant two invitations from one address
// the moment the companion could send email at all.
"quarry-changelog": {"in_app", "push"},
// A test somebody asked for: push ONLY, so checking whether
// push works does not add a row to the inbox every time.
"quarry-test": {"push"},
}
seen := map[string]bool{}
for _, wf := range Workflows() {
steps, ok := want[wf.Name]
if !ok {
t.Fatalf("unexpected workflow %q", wf.Name)
}
seen[wf.Name] = true
if wf.TriggerIdentifier != wf.Name {
t.Fatalf("%s: triggerIdentifier %q must match name", wf.Name, wf.TriggerIdentifier)
}
if len(wf.Steps) != len(steps) {
t.Fatalf("%s: steps = %+v, want %v", wf.Name, wf.Steps, steps)
}
ids := map[string]bool{}
for i, s := range wf.Steps {
if s.Type != steps[i] {
t.Fatalf("%s: step %d is %q, want %q", wf.Name, i, s.Type, steps[i])
}
if s.StepID == "" || ids[s.StepID] {
t.Fatalf("%s: step ids must be present and distinct: %+v", wf.Name, wf.Steps)
}
ids[s.StepID] = true
}
}
for name := range want {
if !seen[name] {
t.Fatalf("workflow %q missing from Workflows()", name)
}
}
}
// A digest's amount must serialize as a JSON NUMBER: notify's
// steptype.sleepConfig unmarshals it into an int64, so a quoted "15"
// makes the step fail to compile and the whole workflow never runs.
func TestDigestAmountIsANumber(t *testing.T) {
for _, wf := range Workflows() {
for _, s := range wf.Steps {
if s.Type != "digest" {
continue
}
b, err := json.Marshal(s.ControlValues)
if err != nil {
t.Fatal(err)
}
var cv struct {
Amount int64 `json:"amount"`
Unit string `json:"unit"`
Type string `json:"type"`
}
if err := json.Unmarshal(b, &cv); err != nil {
t.Fatalf("%s: digest controlValues do not decode as notify reads them: %v (%s)", wf.Name, err, b)
}
// The point is that Amount decoded AS A NUMBER at all: a
// quoted "2" fails notify's unmarshal and the step never
// compiles. The window itself is a default, and lives in
// one place — Workflows.
if cv.Amount != 2 || cv.Unit != "hours" || cv.Type != "regular" {
t.Fatalf("%s: digest window = %+v", wf.Name, cv)
}
}
}
}
// The push steps carry the link as data_json, which notify validates as
// a JSON object whose every value is a string literal (steptype.push).
func TestPushStepsCarryALinkAsStringData(t *testing.T) {
found := 0
for _, wf := range Workflows() {
for _, s := range wf.Steps {
if s.Type != "push" {
continue
}
found++
raw, _ := s.ControlValues["data_json"].(string)
var data map[string]json.RawMessage
if err := json.Unmarshal([]byte(raw), &data); err != nil {
t.Fatalf("%s: data_json is not a JSON object: %q", wf.Name, raw)
}
for k, v := range data {
if len(v) == 0 || v[0] != '"' {
t.Fatalf("%s: data_json[%q] must be a string literal, got %s", wf.Name, k, v)
}
}
if _, ok := data["url"]; !ok {
t.Fatalf("%s: push data carries no url for sw.js to open: %q", wf.Name, raw)
}
}
}
if found != 6 {
t.Fatalf("push steps = %d, want 6 (mention, reply, landing, changelog, test, ops)", found)
}
}