// 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"
"time"
)
// 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
Seen bool
Read bool
CreatedAt string
}
// 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.
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]string `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. It
// deliberately never PUTs over one that is already there: an operator
// may have hand-edited a workflow's copy or channels, and until
// notify#303 ships an idempotent upsert, "leave it alone if present" is
// the only safe default (RFC 0011 §7).
func (c *Client) EnsureWorkflow(ctx context.Context, def WorkflowDef) error {
if c == nil || c.base == "" || def.TriggerIdentifier == "" {
return nil
}
getCtx, cancel := ctxTimeout(ctx)
defer cancel()
req, err := http.NewRequestWithContext(getCtx, "GET",
c.base+"/v1/workflows/"+url.PathEscape(def.TriggerIdentifier), nil)
if err != nil {
return err
}
c.auth(req)
resp, err := c.http.Do(req)
if err != nil {
return fmt.Errorf("notify: ensure workflow %q: %w", def.Name, err)
}
resp.Body.Close()
if resp.StatusCode == http.StatusOK {
return nil // already there — never overwritten (notify#303 tracks upsert)
}
if resp.StatusCode != http.StatusNotFound {
return fmt.Errorf("notify: ensure workflow %q: get status %d", def.Name, resp.StatusCode)
}
postCtx, cancel2 := ctxTimeout(ctx)
defer cancel2()
status, body, err := c.post(postCtx, "/v1/workflows", def)
if err != nil {
return fmt.Errorf("notify: ensure workflow %q: %w", def.Name, err)
}
if status >= 300 && !alreadyExists(status, body) {
return fmt.Errorf("notify: ensure workflow %q: create status %d", def.Name, status)
}
return nil
}