// inbox.go is RFC 0011 §6's half of the thin edge: the calls an inbox
// you can actually CLEAR needs (mark, page, count), the web-push
// credential routes push-with-no-tab rides on, and the channel
// preferences a person sets for themselves.
//
// Every route here was agreed with notify's owner on notify#300 (mark),
// notify#301 (count and pagination), notify#302 (web push) and
// notify#305 (the subscriber-preferences pair and the per-device
// credential DELETE), and every one of them has shipped. Each wire
// assumption still sits behind ONE function, so a route that moves
// costs exactly one body and nothing above it notices.
package notify
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strconv"
"sync"
"time"
)
// PushKeys is the browser's PushSubscription.keys — RFC 8291's two
// halves, which notify's webpush provider encrypts with.
type PushKeys struct {
P256dh string `json:"p256dh"`
Auth string `json:"auth"`
}
// PushSubscription is one browser's push endpoint, exactly as the
// Push API's PushSubscription.toJSON() spells it. quarry does not
// invent this shape and must not reshape it: it is what the browser
// hands the page and what notify hands the push service.
type PushSubscription struct {
Endpoint string `json:"endpoint"`
Keys PushKeys `json:"keys"`
}
// SubscriberPreferences is the body of GET and PUT
// /v1/subscribers/{subscriberId}/preferences, exactly as notify's
// httpapi spells it.
//
// Global reports all five delivery channels every time, an absent one
// meaning enabled. Workflows is keyed by TRIGGER IDENTIFIER — quarry's
// own spelling, not the workflow row id — and carries only the channels
// explicitly overridden for that workflow; an omitted channel inherits
// global. Both sections are PARTIAL on a PUT: notify read-merge-writes,
// so a key left out keeps whatever was stored.
type SubscriberPreferences struct {
Global map[string]bool `json:"global"`
Workflows map[string]map[string]bool `json:"workflows"`
// Digests is the EFFECTIVE batching window for every workflow that
// has a regular digest step, whether or not this person set it:
// Source says which. A sibling of Workflows rather than a member of
// it, so that map stays what every client decodes it as (notify#336).
Digests map[string]DigestWindow `json:"digests,omitempty"`
}
// DigestWindow is how long a digest gathers before it sends, and where
// that number came from.
type DigestWindow struct {
Amount int `json:"amount"`
Unit string `json:"unit"`
// Source is "subscriber" when this person chose it and "workflow"
// when it is the definition's own. A page that cannot tell them
// apart cannot offer "use the default" as a distinct choice.
Source string `json:"source,omitempty"`
}
// Chosen reports whether this window is the subscriber's own.
func (w DigestWindow) Chosen() bool { return w.Source == "subscriber" }
// do runs one request with the client's auth and timeout, returning the
// status and a bounded body. The shared spine under every call below,
// for post's reason: the auth header and the timeout must not drift.
func (c *Client) do(ctx context.Context, method, path string, body any) (int, []byte, error) {
ctx, cancel := ctxTimeout(ctx)
defer cancel()
var r io.Reader
if body != nil {
b, err := json.Marshal(body)
if err != nil {
return 0, nil, err
}
r = bytes.NewReader(b)
}
req, err := http.NewRequestWithContext(ctx, method, c.base+path, r)
if err != nil {
return 0, nil, err
}
if body != nil {
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()
out, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
return resp.StatusCode, out, nil
}
// MarkMessages moves messages to a state: "seen", "read" or "archived".
// Either ids names them or all sweeps the subscriber's whole in-app
// inbox — never both, because notify's bulk route (notify#300) reads
// one or the other and a body carrying both says two things at once.
func (c *Client) MarkMessages(ctx context.Context, subscriber string, ids []string, all bool, as string) error {
if c == nil || c.base == "" || subscriber == "" || as == "" {
return nil
}
if !all && len(ids) == 0 {
return nil
}
body := map[string]any{"subscriberId": subscriber, "as": as}
if all {
body["all"] = true
} else {
body["ids"] = ids
}
status, resp, err := c.do(ctx, "POST", "/v1/messages/mark", body)
if err != nil {
return fmt.Errorf("notify: mark messages: %w", err)
}
if status >= 300 {
return fmt.Errorf("notify: mark messages: status %d: %s", status, bytes.TrimSpace(resp))
}
return nil
}
// messagePage is notify#301's list envelope. Kept unexported and used
// only by decodeMessages, so the one place that knows the wire shape is
// the one place that changes when it moves.
type messagePage struct {
Data []Message `json:"data"`
Next string `json:"next"`
UnreadCount int `json:"unreadCount"`
UnseenCount int `json:"unseenCount"`
}
// decodeMessages reads notify#301's envelope, or the bare array a
// notify older than it answers. The discriminator is the first
// non-whitespace byte, which is the only thing the two bodies disagree
// about at the top level. The bare-array arm is LEGACY: kept so quarry
// still lists an inbox against an older companion, where the counts are
// derived from the page rather than reported.
func decodeMessages(body []byte) (msgs []Message, next string, unread, unseen int, err error) {
trimmed := bytes.TrimLeft(body, " \t\r\n")
if len(trimmed) > 0 && trimmed[0] == '[' {
if err := json.Unmarshal(trimmed, &msgs); err != nil {
return nil, "", 0, 0, err
}
for _, m := range msgs {
if !m.Read {
unread++
}
if !m.Seen {
unseen++
}
}
return msgs, "", unread, unseen, nil
}
var page messagePage
if err := json.Unmarshal(trimmed, &page); err != nil {
return nil, "", 0, 0, err
}
return page.Data, page.Next, page.UnreadCount, page.UnseenCount, nil
}
// Messages is one page of a subscriber's in-app inbox, newest first.
// before is the cursor from a previous page ("" for the first); limit 0
// takes notify's own default. next is "" when the page is the last one.
func (c *Client) Messages(ctx context.Context, subscriber, before string, limit int) ([]Message, string, int, int, error) {
if c == nil || c.base == "" || subscriber == "" {
return nil, "", 0, 0, nil
}
q := url.Values{"subscriberId": {subscriber}, "channel": {"in_app"}}
if limit > 0 {
q.Set("limit", strconv.Itoa(limit))
}
if before != "" {
q.Set("before", before)
}
status, body, err := c.do(ctx, "GET", "/v1/messages?"+q.Encode(), nil)
if err != nil {
return nil, "", 0, 0, fmt.Errorf("notify: messages: %w", err)
}
if status >= 300 {
return nil, "", 0, 0, fmt.Errorf("notify: messages: status %d", status)
}
msgs, next, unread, unseen, err := decodeMessages(body)
if err != nil {
return nil, "", 0, 0, fmt.Errorf("notify: messages: %w", err)
}
return msgs, next, unread, unseen, nil
}
// countBody is notify#301's count answer. Isolated for decodeMessages'
// reason: `count` is an assumed field name until the route ships.
type countBody struct {
Count int `json:"count"`
}
// UnreadCount is how many unread in-app messages a subscriber holds.
// A 404 means notify#301's route is not deployed yet, which is the
// state quarry ships into: fall back to counting a page rather than
// reporting zero, because a badge that reads zero during an outage
// teaches people to stop looking at it.
func (c *Client) UnreadCount(ctx context.Context, subscriber string) (int, error) {
if c == nil || c.base == "" || subscriber == "" {
return 0, nil
}
q := url.Values{"subscriberId": {subscriber}, "channel": {"in_app"}, "unread": {"true"}}
status, body, err := c.do(ctx, "GET", "/v1/messages/count?"+q.Encode(), nil)
if err != nil {
return 0, fmt.Errorf("notify: unread count: %w", err)
}
if status == http.StatusNotFound {
_, _, unread, _, err := c.Messages(ctx, subscriber, "", 0)
return unread, err
}
if status >= 300 {
return 0, fmt.Errorf("notify: unread count: status %d", status)
}
var cb countBody
if err := json.Unmarshal(body, &cb); err != nil {
return 0, fmt.Errorf("notify: unread count: %w", err)
}
return cb.Count, nil
}
// WebPushKey is the VAPID application-server public key a browser needs
// to subscribe (notify#302). quarry never holds the private half — it
// relays the public one so the page can call pushManager.subscribe.
func (c *Client) WebPushKey(ctx context.Context) (string, error) {
if c == nil || c.base == "" {
return "", nil
}
status, body, err := c.do(ctx, "GET", "/v1/integrations/webpush/key", nil)
if err != nil {
return "", fmt.Errorf("notify: webpush key: %w", err)
}
if status >= 300 {
return "", fmt.Errorf("notify: webpush key: status %d", status)
}
var kb struct {
PublicKey string `json:"publicKey"`
}
if err := json.Unmarshal(body, &kb); err != nil {
return "", fmt.Errorf("notify: webpush key: %w", err)
}
return kb.PublicKey, nil
}
// PutWebPush stores one browser's subscription as notify subscriber
// credentials — the same route every other channel's credentials use
// (httpapi.putSubscriberCredentials). Keyed by endpoint on notify's
// side, so a person's several browsers are several rows and re-putting
// the same endpoint replaces rather than duplicates it.
func (c *Client) PutWebPush(ctx context.Context, subscriber string, sub PushSubscription) error {
if c == nil || c.base == "" || subscriber == "" || sub.Endpoint == "" {
return nil
}
status, body, err := c.do(ctx, "PUT", "/v1/subscribers/"+url.PathEscape(subscriber)+"/credentials",
map[string]any{"providerId": "webpush", "credentials": sub})
if err != nil {
return fmt.Errorf("notify: put webpush: %w", err)
}
if status >= 300 {
return fmt.Errorf("notify: put webpush: status %d: %s", status, bytes.TrimSpace(body))
}
return nil
}
// DeleteWebPush drops one endpoint's credential (204). 404 and 410 are
// success too: the end state unsubscribe asked for — that endpoint no
// longer receives — is exactly what a credential notify does not hold,
// or one the push service already discarded, already is. Anything else
// is an error, and the caller keeps its row and retries.
func (c *Client) DeleteWebPush(ctx context.Context, subscriber, endpoint string) error {
if c == nil || c.base == "" || subscriber == "" || endpoint == "" {
return nil
}
path := "/v1/subscribers/" + url.PathEscape(subscriber) + "/credentials/webpush?" +
url.Values{"endpoint": {endpoint}}.Encode()
status, _, err := c.do(ctx, "DELETE", path, nil)
if err != nil {
return fmt.Errorf("notify: delete webpush: %w", err)
}
if status >= 300 && status != http.StatusNotFound && status != http.StatusGone {
return fmt.Errorf("notify: delete webpush: status %d", status)
}
return nil
}
// Preferences reads a subscriber's stored channel preferences.
//
// A 404 or 405 is NOT an empty read: it says notify did not answer at
// all — no such subscriber there, or something else mounted at the
// path — and a caller that cannot tell that from "nothing is
// overridden" would show somebody quarry's defaults as if they were
// notify's stored settings.
func (c *Client) Preferences(ctx context.Context, subscriber string) (SubscriberPreferences, error) {
var out SubscriberPreferences
if c == nil || c.base == "" || subscriber == "" {
return out, nil
}
status, body, err := c.do(ctx, "GET", "/v1/subscribers/"+url.PathEscape(subscriber)+"/preferences", nil)
if err != nil {
return out, fmt.Errorf("notify: preferences: %w", err)
}
if routeMissing(status) {
return out, fmt.Errorf("notify: preferences: status %d: %w", status, ErrRouteMissing)
}
if status >= 300 {
return out, fmt.Errorf("notify: preferences: status %d", status)
}
if err := json.Unmarshal(body, &out); err != nil {
return SubscriberPreferences{}, fmt.Errorf("notify: preferences: %w", err)
}
return out, nil
}
// SetPreferences writes the per-workflow overrides, keyed by trigger
// identifier. Every cell of quarry's matrix is sent EXPLICITLY, true as
// well as false: notify merges what it is given and reads an absent
// channel as enabled, so an unticked box that arrived as absence would
// be a box that never took.
//
// The global section is deliberately never written. It is notify's own
// surface, quarry's matrix has no cell for it, and a client that posted
// it would rewrite a channel somebody set somewhere quarry cannot even
// show them.
// digests carries the batching windows: a value sets one, and a nil
// value clears it back to the workflow's own. notify MERGES what it is
// given here, so a workflow absent from the map keeps whatever it had —
// the opposite of the channel matrix above, and deliberately so: a
// window nobody touched is not a window somebody cleared.
func (c *Client) SetPreferences(ctx context.Context, subscriber string,
workflows map[string]map[string]bool, digests map[string]*DigestWindow) error {
if c == nil || c.base == "" || subscriber == "" {
return nil
}
if workflows == nil {
workflows = map[string]map[string]bool{}
}
body := map[string]any{"workflows": workflows}
if len(digests) > 0 {
body["digests"] = digests
}
status, respBody, err := c.do(ctx, "PUT", "/v1/subscribers/"+url.PathEscape(subscriber)+"/preferences",
body)
if err != nil {
return fmt.Errorf("notify: set preferences: %w", err)
}
if routeMissing(status) {
return fmt.Errorf("notify: set preferences: status %d: %w", status, ErrRouteMissing)
}
if status >= 300 {
return fmt.Errorf("notify: set preferences: status %d: %s", status, bytes.TrimSpace(respBody))
}
return nil
}
// ErrRouteMissing marks a preferences call notify did not answer: a 404
// (no such subscriber there, or nothing mounted at the path) or a 405.
// It is deliberately NOT an ordinary failure — a caller that cannot
// tell "notify said nothing" from "notify said nothing is overridden"
// shows quarry's own defaults as somebody's saved settings.
var ErrRouteMissing = errors.New("notify: route not deployed")
// routeMissing reports whether status says "this route is not here":
// 404 when nothing is mounted at the path, 405 when something is but
// not for this method (a PUT against a GET-only preferences route).
func routeMissing(status int) bool {
return status == http.StatusNotFound || status == http.StatusMethodNotAllowed
}
// WorkflowInfo is one row of GET /v1/workflows (notify's workflowDTO).
// The two fields that matter here are the pair nothing else relates: ID
// is the workflow ULID a Message.WorkflowID carries, TriggerIdentifier
// is the slug quarry named the workflow by and triggers it with.
type WorkflowInfo struct {
ID string `json:"id"`
Name string `json:"name"`
TriggerIdentifier string `json:"triggerIdentifier"`
}
// Workflows lists the workflows in notify's scope. Reads either the
// bare array notify serves today or an envelope, by decodeMessages'
// discriminator and for its reason.
func (c *Client) Workflows(ctx context.Context) ([]WorkflowInfo, error) {
if c == nil || c.base == "" {
return nil, nil
}
status, body, err := c.do(ctx, "GET", "/v1/workflows", nil)
if err != nil {
return nil, fmt.Errorf("notify: workflows: %w", err)
}
if status >= 300 {
return nil, fmt.Errorf("notify: workflows: status %d", status)
}
trimmed := bytes.TrimLeft(body, " \t\r\n")
if len(trimmed) > 0 && trimmed[0] == '[' {
var out []WorkflowInfo
if err := json.Unmarshal(trimmed, &out); err != nil {
return nil, fmt.Errorf("notify: workflows: %w", err)
}
return out, nil
}
var env struct {
Data []WorkflowInfo `json:"data"`
}
if err := json.Unmarshal(trimmed, &env); err != nil {
return nil, fmt.Errorf("notify: workflows: %w", err)
}
return env.Data, nil
}
// workflowNamesTTL is how long a resolved map is trusted, and also the
// floor between two refresh ATTEMPTS. Short because a workflow created
// while quarry is running must become nameable without a restart; not
// shorter, because every unresolved id would otherwise re-ask notify
// and a page of them would be a page of round trips.
const workflowNamesTTL = time.Minute
// workflowNamesMissGap is the floor between two refreshes triggered by
// a LOOKUP MISS rather than by staleness: a workflow created since the
// last load should become nameable in seconds, but a page full of ids
// notify has never heard of must still cost one call, not one each.
const workflowNamesMissGap = 5 * time.Second
// WorkflowNames turns the workflow ULID a Message carries into the
// trigger identifier quarry knows it by, which is what makes an inbox
// row's reason quarry's own word rather than an opaque id.
//
// It is now a FALLBACK: notify puts the identifier on the message
// (Message.WorkflowIdentifier), and Identify only reaches for the map
// on rows that carry none — an older companion, or a workflow deleted
// since the message was sent.
//
// Never blocks a page on notify: an unresolvable id resolves to itself,
// and a failed refresh is not retried until the TTL is up. The caller
// renders a generic reason and the inbox still lists.
type WorkflowNames struct {
c *Client
ttl time.Duration
mu sync.Mutex
byID map[string]string
byName map[string]string
lastTry time.Time // any attempt, successful or not — the rate limit
lastOK time.Time // the last attempt that actually loaded a map
}
// NewWorkflowNames returns a map over c. A nil or disabled client is
// allowed: every lookup then falls back to the id, which is what a
// forge with no notify would show anyway.
func NewWorkflowNames(c *Client) *WorkflowNames {
return &WorkflowNames{c: c, ttl: workflowNamesTTL}
}
// beginRefresh decides whether a load is due and, if it is, claims the
// attempt. Caller holds mu; the HTTP call happens OUTSIDE it (see
// refresh), because holding a mutex across a five-second round trip
// would make every concurrent page wait on the slowest one.
func (w *WorkflowNames) beginRefresh(force bool) bool {
if w.c == nil {
return false
}
now := time.Now()
gap := w.ttl
if force {
gap = workflowNamesMissGap
}
if !w.lastTry.IsZero() && now.Sub(w.lastTry) < gap {
return false
}
if !force && !w.lastOK.IsZero() && now.Sub(w.lastOK) < w.ttl {
return false
}
w.lastTry = now
return true
}
// refresh reloads the map when it is stale, or when force says a lookup
// missed. Both are rate-limited by lastTry, so a notify outage costs one
// call per window rather than one per row, and a failure leaves the
// previously resolved names in place.
//
// Called with mu HELD; it releases mu for the request and retakes it to
// install the result. Callers must therefore re-read anything they
// looked at before calling.
func (w *WorkflowNames) refresh(ctx context.Context, force bool) {
if w == nil || !w.beginRefresh(force) {
return
}
w.mu.Unlock()
wfs, err := w.c.Workflows(ctx)
w.mu.Lock()
if err != nil {
return
}
byID := make(map[string]string, len(wfs))
byName := make(map[string]string, len(wfs))
for _, wf := range wfs {
if wf.ID == "" || wf.TriggerIdentifier == "" {
continue
}
byID[wf.ID] = wf.TriggerIdentifier
byName[wf.TriggerIdentifier] = wf.ID
}
w.byID, w.byName, w.lastOK = byID, byName, time.Now()
}
// Resolved reports whether the map has ever loaded. A caller that finds
// no stored row for a workflow needs this to tell "there is no such
// preference" from "quarry could not work out what to look under" —
// the two answers mean opposite things to somebody reading a settings
// page, and only one of them is safe to render as a default.
func (w *WorkflowNames) Resolved(ctx context.Context) bool {
if w == nil || w.c == nil {
return false
}
w.mu.Lock()
defer w.mu.Unlock()
w.refresh(ctx, false)
return !w.lastOK.IsZero()
}
// Identifiers resolves a whole page's workflow ids at once, costing at
// most one refresh however many of them are unknown. An id that cannot
// be resolved maps to ITSELF — the caller's fallback is then visible in
// the row rather than hidden behind a blank reason.
func (w *WorkflowNames) Identifiers(ctx context.Context, ids []string) map[string]string {
out := make(map[string]string, len(ids))
if w == nil {
for _, id := range ids {
out[id] = id
}
return out
}
w.mu.Lock()
defer w.mu.Unlock()
w.refresh(ctx, false)
unknown := false
for _, id := range ids {
if id != "" && w.byID[id] == "" {
unknown = true
break
}
}
if unknown {
// One retry for the page: a workflow created since the last
// refresh is exactly the case the TTL alone would hide.
w.refresh(ctx, true)
}
for _, id := range ids {
if name := w.byID[id]; name != "" {
out[id] = name
continue
}
out[id] = id
}
return out
}
// Identify resolves a whole page's reasons: the trigger identifier for
// each message, keyed by MESSAGE id.
//
// notify now puts the identifier on the message itself, so a page from
// a current companion costs NO workflow lookup at all — which is the
// point of the field, and the difference between listing an inbox and
// listing an inbox plus a round trip. The map is consulted only for
// rows that carry none: an older companion, or a message whose workflow
// has since been deleted.
func (w *WorkflowNames) Identify(ctx context.Context, msgs []Message) map[string]string {
out := make(map[string]string, len(msgs))
var unnamed []string
for _, m := range msgs {
if m.WorkflowIdentifier != "" {
out[m.ID] = m.WorkflowIdentifier
continue
}
unnamed = append(unnamed, m.WorkflowID)
}
if len(unnamed) == 0 {
return out
}
byID := w.Identifiers(ctx, unnamed)
for _, m := range msgs {
if m.WorkflowIdentifier == "" {
out[m.ID] = byID[m.WorkflowID]
}
}
return out
}
// Identifier is Identifiers for one id.
func (w *WorkflowNames) Identifier(ctx context.Context, id string) string {
return w.Identifiers(ctx, []string{id})[id]
}
// ID is the reverse: the workflow ULID for a trigger identifier, or ""
// when it cannot be resolved. Used to key a payload by the id a route
// expects without making the caller guess one.
func (w *WorkflowNames) ID(ctx context.Context, identifier string) string {
if w == nil || identifier == "" {
return ""
}
w.mu.Lock()
defer w.mu.Unlock()
w.refresh(ctx, false)
if id := w.byName[identifier]; id != "" {
return id
}
w.refresh(ctx, true)
return w.byName[identifier]
}