// 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.
// The address is sent on EVERY call, not only the first: notify's
// subscriber upsert REPLACES the row, so a call that omits the email
// blanks the one already there — and a subscriber with no address is
// one no email can reach, silently. Callers that do not know the
// address must not be able to reach this function; see provisionAdapter.
func (c *Client) EnsureSubscriber(ctx context.Context, id, email 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, "email": email})
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
}
// EmailSender is a mail provider notify should deliver email through,
// as quarry knows it: the same credential quarry already holds for its
// own invite mail.
type EmailSender struct {
// Provider is notify's own spelling — "resend", "ses",
// "cloudflare-email" or "" for SMTP.
Provider string
// Identifier names the integration in notify's console.
Identifier string
// From is the sender address, and Key the provider credential.
From, Key string
}
// EnsureEmailIntegration gives notify a way to send email, if it has
// none.
//
// Without this a self-hoster has to POST a credential into the
// companion's API by hand before any email works, and nothing on either
// side says that is missing — quarry's settings page offers an email
// column, notify accepts the preference, and every email errors with
// "store: not found" where only an operator reading the database would
// find it. quarry already holds a mail credential for invites; this
// hands notify the same one.
//
// Existing integrations are LEFT ALONE. An operator who configured a
// different provider in notify's console meant it, and a forge restart
// must not replace their choice.
func (c *Client) EnsureEmailIntegration(ctx context.Context, s EmailSender) (created bool, err error) {
if c == nil || c.base == "" || s.Key == "" || s.From == "" {
return false, nil
}
ctx, cancel := ctxTimeout(ctx)
defer cancel()
status, body, err := c.do(ctx, "GET", "/v1/integrations", nil)
if err != nil {
return false, fmt.Errorf("notify: integrations: %w", err)
}
if status >= 300 {
return false, fmt.Errorf("notify: integrations: status %d", status)
}
var rows []struct {
Channel string `json:"channel"`
}
if err := json.Unmarshal(body, &rows); err != nil {
return false, fmt.Errorf("notify: integrations: %w", err)
}
for _, r := range rows {
if r.Channel == "email" {
return false, nil
}
}
id := s.Identifier
if id == "" {
id = "quarry-mail"
}
status, body, err = c.post(ctx, "/v1/integrations/email", map[string]any{
"provider": s.Provider, "identifier": id, "from": s.From,
"apiKey": s.Key, "primary": true,
})
if err != nil {
return false, fmt.Errorf("notify: email integration: %w", err)
}
if status >= 300 && !alreadyExists(status, body) {
return false, fmt.Errorf("notify: email integration: status %d: %s", status, clipBody(body))
}
return true, 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"`
}
// WorkflowOutcome is what EnsureWorkflow's upsert did to the stored
// workflow, as notify reports it.
type WorkflowOutcome string
const (
WorkflowCreated WorkflowOutcome = "created"
WorkflowUnchanged WorkflowOutcome = "unchanged"
WorkflowUpdated WorkflowOutcome = "updated"
)
// EnsureWorkflow makes notify's workflow of def's slug match def,
// through notify#303's idempotent upsert: one
// PUT /v1/workflows/{triggerIdentifier}?onConflict=update. notify
// compares the definition (name, steps) canonically and creates it when
// absent, writes nothing when it already matches, and rewrites it IN
// PLACE when it differs — same workflow id, so subscribers'
// per-workflow preferences, which notify keys by that id, survive.
// Never delete and recreate a workflow to change it.
//
// quarry owns the definitions in Workflows(): a step or template it
// changes reaches an existing forge on the next start, and a hand edit
// to one of those workflows in notify's console is undone by it. Only
// the slugs passed here are touched.
func (c *Client) EnsureWorkflow(ctx context.Context, def WorkflowDef) (WorkflowOutcome, 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=update",
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))
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
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))
}
var answer struct {
Outcome WorkflowOutcome `json:"outcome"`
}
json.Unmarshal(body, &answer)
switch {
case answer.Outcome != "":
return answer.Outcome, nil
case resp.StatusCode == http.StatusCreated:
return WorkflowCreated, nil
default:
return WorkflowUnchanged, nil
}
}