// Package notify is quarry's thin edge to the notify companion (0001
// §5/§8): quarry does not build notification logic, it fires triggers.
// A trigger is the Novu-compatible POST /v1/events/trigger the notify
// RFC preserves as its public contract; notify resolves recipients,
// runs the workflow DAG, and owns channels, digests, and the inbox.
// Triggers are best-effort — a notify outage must never fail the user
// action that raised it.
package notify
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"time"
"craftdesign.group/quarry/internal/fault"
)
// Client posts triggers to a notify base URL. A nil Client (base "") is
// the disabled no-op: Trigger returns nil without a call.
type Client struct {
base string
key string
http *http.Client
}
// New returns a client for baseURL, or a disabled client when baseURL
// is empty (notify not configured). apiKey is notify's console key: its
// /v1 routes sit behind an API-key middleware, so a client without one
// is answered 401 on every call.
func New(baseURL, apiKey string) *Client {
return &Client{base: baseURL, key: apiKey, http: &http.Client{Timeout: 5 * time.Second}}
}
// auth applies notify's scheme. Named rather than inlined so the two
// callers cannot drift.
func (c *Client) auth(req *http.Request) {
if c.key != "" {
req.Header.Set("Authorization", "ApiKey "+c.key)
}
}
// Message is one in-app notification as notify serves it.
//
// The field names are notify's wire names, which are its Go field names:
// domain.Message carries db tags but no json tags, so the encoder falls
// back to the identifiers. Renaming a field there renames it here.
type Message struct {
ID string
Subject string
Content string
// WorkflowID is the ROW id of the workflow that sent it — a ULID,
// meaningless to a reader. Kept because it is what an older
// companion offers; WorkflowIdentifier is what a reason is made of.
WorkflowID string
// WorkflowIdentifier is that workflow's TRIGGER IDENTIFIER —
// "quarry-mention", quarry's own spelling — resolved by notify at
// read time (its domain.Message, notify#305). Empty on an older
// companion and on a message whose workflow has since been deleted;
// WorkflowNames is the fallback for those and only those.
WorkflowIdentifier string
// TransactionID is the idempotency key the trigger carried. quarry
// sets one on every person-addressed trigger and writes the same
// key onto its own attention event, so this is what pairs the two
// records of one mention when `inbox.list` merges them. Empty on a
// message notify raised for anything else.
TransactionID string
Seen bool
Read bool
Archived bool
CreatedAt string
// Extras is notify's in_app rich-field snapshot (avatar, actions,
// redirect, data), the one field on domain.Message carrying an
// explicit json tag. Held raw because quarry needs one thing out of
// it — the redirect — and decoding the rest would pin a shape that
// is notify's to change.
Extras json.RawMessage `json:"extras"`
}
// Link is the message's redirect URL, empty when it carries none. The
// one thing quarry reads out of Extras: an inbox row that cannot be
// clicked through to the thing it is about is a notification that
// makes somebody go looking.
func (m Message) Link() string {
if len(m.Extras) == 0 {
return ""
}
var e struct {
Redirect struct {
URL string `json:"url"`
} `json:"redirect"`
}
if err := json.Unmarshal(m.Extras, &e); err != nil {
return ""
}
// notify snapshots a step's rich fields verbatim: subject and body
// are rendered against the payload, redirect is not (notify#328).
// An unrendered template is not a destination — a row linking to
// "{{payload.link}}" is worse than a row that does not link at all.
if strings.Contains(e.Redirect.URL, "{{") {
return ""
}
return e.Redirect.URL
}
// Inbox returns a subscriber's in-app messages, newest first. notify
// owns the notification model — quarry only displays it, which is what
// "embed the inbox" means (0001 §5).
func (c *Client) Inbox(ctx context.Context, subscriberID string) ([]Message, error) {
if c == nil || c.base == "" || subscriberID == "" {
return nil, nil
}
u := c.base + "/v1/messages?channel=in_app&subscriberId=" + url.QueryEscape(subscriberID)
req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
if err != nil {
return nil, err
}
c.auth(req)
resp, err := c.http.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return nil, fmt.Errorf("notify: inbox: status %d", resp.StatusCode)
}
var msgs []Message
if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&msgs); err != nil {
return nil, fmt.Errorf("notify: inbox: %w", err)
}
return msgs, nil
}
// Recipient is a trigger's "to". Exactly one addressing mode applies:
// SubscriberID names one person, TopicKey fans out to every member of a
// topic. They are DIFFERENT fields in notify's API — putting a topic key
// in SubscriberID does not multicast, it silently invents a subscriber
// with that name, and the people who should have been notified are not.
type Recipient struct {
SubscriberID string `json:"subscriberId,omitempty"`
Email string `json:"email,omitempty"`
TopicKey string `json:"topicKey,omitempty"`
}
type triggerBody struct {
Name string `json:"name"`
To Recipient `json:"to"`
Payload map[string]any `json:"payload"`
TransactionID string `json:"transactionId,omitempty"`
}
// Trigger starts the named workflow for one recipient. transactionID is
// the idempotency key (notify maps it to dagnats WithIdempotencyKey), so
// a replayed or duplicated event notifies at most once. A disabled
// client no-ops.
func (c *Client) Trigger(ctx context.Context, name string, to Recipient, payload map[string]any, transactionID string) error {
if c == nil || c.base == "" {
return nil
}
body, err := json.Marshal(triggerBody{Name: name, To: to, Payload: payload, TransactionID: transactionID})
if err != nil {
return err
}
req, err := http.NewRequestWithContext(ctx, "POST", c.base+"/v1/events/trigger", bytes.NewReader(body))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
c.auth(req)
resp, err := c.http.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return fmt.Errorf("notify: trigger %q: status %d", name, resp.StatusCode)
}
return nil
}
// post is the shared shape behind every "create this, and a prior
// create is success" call below: notify's create routes answer 409 (or
// occasionally 400 with an "already exists" body) when the row is
// already there, and every one of these calls is idempotent by intent
// — provisioning runs on every account.add, every grant, every
// startup, and none of it should treat "it's already provisioned" as a
// failure.
func (c *Client) post(ctx context.Context, path string, body any) (status int, respBody []byte, err error) {
b, err := json.Marshal(body)
if err != nil {
return 0, nil, err
}
req, err := http.NewRequestWithContext(ctx, "POST", c.base+path, bytes.NewReader(b))
if err != nil {
return 0, nil, err
}
req.Header.Set("Content-Type", "application/json")
c.auth(req)
resp, err := c.http.Do(req)
if err != nil {
return 0, nil, err
}
defer resp.Body.Close()
respBody, _ = io.ReadAll(io.LimitReader(resp.Body, 64<<10))
return resp.StatusCode, respBody, nil
}
// alreadyExists reports whether a create's non-2xx answer means "this
// row is already there" rather than a real failure: notify answers a
// duplicate create with 409, and some routes say 400 with an
// "already exists" body instead. Either way the row this call wanted
// now exists, which is what every caller here is actually asking for.
// clipBody keeps enough of a refusal to diagnose it and no more: the
// first line is where a service says what it objected to.
func clipBody(b []byte) string {
line, _, _ := strings.Cut(string(b), "\n")
line = strings.TrimSpace(line)
if len(line) > 200 {
line = line[:200] + "…"
}
return line
}
func alreadyExists(status int, body []byte) bool {
if status == http.StatusConflict {
return true
}
return status == http.StatusBadRequest && bytes.Contains(bytes.ToLower(body), []byte("already exist"))
}
func ctxTimeout(ctx context.Context) (context.Context, context.CancelFunc) {
return context.WithTimeout(ctx, 5*time.Second)
}
// EnsureSubscriber creates a notify subscriber for id if one is not
// already there. Best-effort by construction: the caller (account.add)
// must never fail because notify is unreachable.
func (c *Client) EnsureSubscriber(ctx context.Context, id string) error {
if c == nil || c.base == "" || id == "" {
return nil
}
ctx, cancel := ctxTimeout(ctx)
defer cancel()
status, body, err := c.post(ctx, "/v1/subscribers", map[string]string{"subscriberId": id})
if err != nil {
return fmt.Errorf("notify: ensure subscriber %q: %w", id, err)
}
if status >= 300 && !alreadyExists(status, body) {
return fmt.Errorf("notify: ensure subscriber %q: status %d", id, status)
}
return nil
}
// EnsureTopic creates a topic (key, name) if one is not already there.
func (c *Client) EnsureTopic(ctx context.Context, key, name string) error {
if c == nil || c.base == "" || key == "" {
return nil
}
ctx, cancel := ctxTimeout(ctx)
defer cancel()
status, body, err := c.post(ctx, "/v1/topics", map[string]string{"key": key, "name": name})
if err != nil {
return fmt.Errorf("notify: ensure topic %q: %w", key, err)
}
if status >= 300 && !alreadyExists(status, body) {
return fmt.Errorf("notify: ensure topic %q: status %d", key, status)
}
return nil
}
// AddTopicMember adds subscriber to topic key. Adding an existing
// member is notify's own no-op (not ours to detect), so any success
// status is success and any failure is reported as one.
func (c *Client) AddTopicMember(ctx context.Context, key, subscriber string) error {
if c == nil || c.base == "" || key == "" || subscriber == "" {
return nil
}
ctx, cancel := ctxTimeout(ctx)
defer cancel()
status, body, err := c.post(ctx, "/v1/topics/"+url.PathEscape(key)+"/subscribers",
map[string]string{"subscriberId": subscriber})
if err != nil {
return fmt.Errorf("notify: add topic member %q/%q: %w", key, subscriber, err)
}
if status >= 300 && !alreadyExists(status, body) {
return fmt.Errorf("notify: add topic member %q/%q: status %d", key, subscriber, status)
}
return nil
}
// RemoveTopicMember removes subscriber from topic key (account.revoke).
// Removing an absent member is treated as success — the end state
// (not a member) is what revoke asked for.
func (c *Client) RemoveTopicMember(ctx context.Context, key, subscriber string) error {
if c == nil || c.base == "" || key == "" || subscriber == "" {
return nil
}
ctx, cancel := ctxTimeout(ctx)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "DELETE",
c.base+"/v1/topics/"+url.PathEscape(key)+"/subscribers/"+url.PathEscape(subscriber), nil)
if err != nil {
return err
}
c.auth(req)
resp, err := c.http.Do(req)
if err != nil {
return fmt.Errorf("notify: remove topic member %q/%q: %w", key, subscriber, err)
}
defer resp.Body.Close()
if resp.StatusCode >= 300 && resp.StatusCode != http.StatusNotFound {
return fmt.Errorf("notify: remove topic member %q/%q: status %d", key, subscriber, resp.StatusCode)
}
return nil
}
// WorkflowStep is one step of a workflow definition (EnsureWorkflow).
type WorkflowStep struct {
StepID string `json:"stepId"`
Type string `json:"type"`
ControlValues map[string]any `json:"controlValues"`
}
// WorkflowDef is a workflow as EnsureWorkflow and notify's
// POST /v1/workflows both understand it (docs/running.md's curl shape,
// now a Go value in workflows.go instead of a paragraph of curl).
type WorkflowDef struct {
Name string `json:"name"`
TriggerIdentifier string `json:"triggerIdentifier"`
Steps []WorkflowStep `json:"steps"`
}
// EnsureWorkflow creates def if no workflow of that slug exists yet,
// through notify#303's idempotent upsert: one
// PUT /v1/workflows/{triggerIdentifier}?onConflict=keep, which creates
// when the slug is absent and leaves an existing definition exactly as
// it is.
//
// `keep` and never `update`: an operator may have hand-edited a
// workflow's copy or its channels, and provisioning runs on every
// startup — an upsert that overwrote would undo that edit on the next
// restart, silently (RFC 0011 §7).
func (c *Client) EnsureWorkflow(ctx context.Context, def WorkflowDef) error {
if c == nil || c.base == "" || def.TriggerIdentifier == "" {
return nil
}
ctx, cancel := ctxTimeout(ctx)
defer cancel()
b, err := json.Marshal(def)
if err != nil {
return err
}
req, err := http.NewRequestWithContext(ctx, "PUT",
c.base+"/v1/workflows/"+url.PathEscape(def.TriggerIdentifier)+"?onConflict=keep",
bytes.NewReader(b))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
c.auth(req)
resp, err := c.http.Do(req)
if err != nil {
// The workflow's name is a FIELD, not part of the sentence:
// six workflows failing on one unreachable notify is one fault,
// and a name in the message made it six rows in the journal.
return fault.Wrap(err, "notify workflow could not be provisioned").
Cost("the alerts this workflow carries are not delivered; nothing else is affected").
Do("check that notify is reachable — the workflows are re-provisioned on the next start").
With("workflow", def.Name, "endpoint", c.base)
}
defer resp.Body.Close()
body, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
// 200 is "it was already there and was kept", 201 "it was created".
// Both are the end state this call asks for.
if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusCreated {
return nil
}
if alreadyExists(resp.StatusCode, body) {
return nil
}
return fault.New("notify refused a workflow").
Cost("the alerts this workflow carries are not delivered; nothing else is affected").
With("workflow", def.Name, "endpoint", c.base, "status", resp.StatusCode,
"body", clipBody(body))
}