// 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) and notify#302 (web push); the
// preferences pair is read off notify's own internal/preferences
// service and domain.Preference, which has no HTTP route yet. That is
// why each assumption sits behind ONE function: when a route lands
// differently, exactly one body changes 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"`
}
// Preference is one stored channel-preference row, mirroring notify's
// domain.Preference: a subscriber's global toggles (Type
// "subscriberGlobal", no workflow) or one workflow's overrides (Type
// "workflow"). Absent channel => enabled, which is notify's own
// resolution rule (internal/preferences: "a channel is enabled unless a
// level explicitly sets it false").
type Preference struct {
Type string `json:"type"`
WorkflowID string `json:"workflowId,omitempty"`
Channels map[string]bool `json:"channels"`
}
// PreferenceWorkflow is Preference.Type for a per-workflow row.
const PreferenceWorkflow = "workflow"
// 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 either notify#301's envelope or the bare array
// /v1/messages answers today. The discriminator is the first
// non-whitespace byte, which is the only thing the two bodies disagree
// about at the top level. On the bare array the counts are derived from
// the page — honest for a first page and the best available until the
// envelope lands.
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. 404 and 410 are
// success: the end state unsubscribe asked for — that endpoint no
// longer receives — is exactly what a gone endpoint already is.
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 routeMissing(status) {
// notify#305: the credential DELETE is not mounted yet, so a
// 404 here does NOT mean "that endpoint is already gone" — it
// means nobody was asked. The caller must keep its row and try
// again rather than forget a device notify still pushes to.
return fmt.Errorf("notify: delete webpush: status %d: %w", status, ErrRouteMissing)
}
if status >= 300 && status != http.StatusGone {
return fmt.Errorf("notify: delete webpush: status %d", status)
}
return nil
}
// decodePreferences reads either an envelope ({"data":[...]}) or a bare
// array, for decodeMessages' reason and by the same discriminator.
func decodePreferences(body []byte) ([]Preference, error) {
trimmed := bytes.TrimLeft(body, " \t\r\n")
if len(trimmed) > 0 && trimmed[0] == '[' {
var out []Preference
return out, json.Unmarshal(trimmed, &out)
}
var env struct {
Data []Preference `json:"data"`
}
if err := json.Unmarshal(trimmed, &env); err != nil {
return nil, err
}
return env.Data, nil
}
// Preferences reads a subscriber's stored channel preferences. A
// subscriber who has never saved one is not an error — notify holds no
// row and every channel is on, which is what an empty slice means here.
func (c *Client) Preferences(ctx context.Context, subscriber string) ([]Preference, error) {
if c == nil || c.base == "" || subscriber == "" {
return nil, nil
}
status, body, err := c.do(ctx, "GET", "/v1/subscribers/"+url.PathEscape(subscriber)+"/preferences", nil)
if err != nil {
return nil, fmt.Errorf("notify: preferences: %w", err)
}
if status == http.StatusNotFound {
return nil, nil
}
if status >= 300 {
return nil, fmt.Errorf("notify: preferences: status %d", status)
}
prefs, err := decodePreferences(body)
if err != nil {
return nil, fmt.Errorf("notify: preferences: %w", err)
}
return prefs, nil
}
// SetPreferences writes a subscriber's preference rows. The whole set
// is sent, not a delta: a matrix page shows every cell, so what it
// posts IS the whole intended state and a partial write would leave
// cells the person just unticked still on.
func (c *Client) SetPreferences(ctx context.Context, subscriber string, prefs []Preference) error {
if c == nil || c.base == "" || subscriber == "" {
return nil
}
if prefs == nil {
prefs = []Preference{}
}
status, body, err := c.do(ctx, "PUT", "/v1/subscribers/"+url.PathEscape(subscriber)+"/preferences",
map[string]any{"preferences": prefs})
if err != nil {
return fmt.Errorf("notify: set preferences: %w", err)
}
if routeMissing(status) {
// notify#305: there is no PUT /v1/subscribers/{id}/preferences
// yet. The caller writes its forge-local half first and reports
// that this half did not land, rather than failing an operation
// notify has no say in.
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(body))
}
return nil
}
// ErrRouteMissing marks a route quarry needs that this notify does not
// serve yet: a 404 or 405 on one of the surfaces agreed but not shipped
// (the subscriber-preferences pair and the webpush credential DELETE,
// notify#305). It is deliberately NOT an ordinary failure — a caller
// that cannot tell "not yet" from "broken" must either fail an
// operation notify has no say in, or swallow a real outage.
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 the ONLY thing that
// makes an inbox row's reason quarry's own word rather than an opaque
// id. notify's Message.WorkflowID is the workflow ROW id — quarry's
// spelling ("quarry-mention") appears nowhere on a message, so without
// this map a reason derived from it is either blank or, worse, quarry's
// own guess echoed back at itself.
//
// 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}
}
// 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. Caller holds mu.
func (w *WorkflowNames) refresh(ctx context.Context, force bool) {
if w == nil || w.c == nil {
return
}
now := time.Now()
gap := w.ttl
if force {
gap = workflowNamesMissGap
}
if !w.lastTry.IsZero() && now.Sub(w.lastTry) < gap {
return
}
if !force && !w.lastOK.IsZero() && now.Sub(w.lastOK) < w.ttl {
return
}
w.lastTry = now
wfs, err := w.c.Workflows(ctx)
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, now
}
// 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
}
// 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]
}