Last updated 2026-05-28

gRPC / Connect API Reference

Two Connect services live in proto/notify/v1. All RPCs accept JSON over HTTP/1 + HTTP/2 (Connect) and binary gRPC — the wire format is negotiated by the Content-Type header. Field numbers are frozen forever; additive fields land at the end with a new number, never reuse, never renumber.

Quick navigation

Enums — DeliveryChannel, DeliveryStatus, DeviceType
NotificationInternalService.Notify — backend producers
NotificationClientService — browser / mobile recipients
Error codes

Enums

DeliveryChannel

enum DeliveryChannel {
  DELIVERY_CHANNEL_UNSPECIFIED = 0;
  DELIVERY_CHANNEL_IN_APP      = 1;
  DELIVERY_CHANNEL_EMAIL       = 2;
  DELIVERY_CHANNEL_WEB_PUSH    = 3;
  DELIVERY_CHANNEL_MOBILE_PUSH = 4;
  DELIVERY_CHANNEL_SMS         = 5;
  DELIVERY_CHANNEL_WHATSAPP    = 6;
}

The string form used in NotifyRequest.addresses.byChannel keys is the lowercase suffix: in_app, email, web_push, mobile_push, sms, whatsapp. (In-app needs no entry — destination is the user id itself.)

DeliveryStatus

enum DeliveryStatus {
  DELIVERY_STATUS_UNSPECIFIED = 0;
  DELIVERY_STATUS_PENDING     = 1;  // stored, not yet handed to a provider
  DELIVERY_STATUS_DELIVERED   = 2;  // provider accepted the message
  DELIVERY_STATUS_ACKED       = 3;  // transport-level confirmation (e.g. push receipt)
  DELIVERY_STATUS_READ        = 4;  // recipient-driven
  DELIVERY_STATUS_FAILED      = 5;  // provider rejection
}

DeviceType

enum DeviceType {
  DEVICE_TYPE_UNSPECIFIED = 0;
  DEVICE_TYPE_BROWSER     = 1;
  DEVICE_TYPE_ANDROID     = 2;
  DEVICE_TYPE_IOS         = 3;
}

The platform supports at most one registered token per (user, device_type). Re-registering rotates the token in place; it does not create a new device row.

NotificationInternalService

Backend-to-backend surface. Speaks gRPC and Connect. Auth is X-Notify-Internal-Token at the transport layer.

Notify

service NotificationInternalService {
  // Creates per-user notification rows and attempts immediate delivery
  // for every (user, channel) where a provider+destination is configured.
  // Storage is idempotent on (tenant_id, user_id, notification_id).
  rpc Notify(NotifyRequest) returns (NotifyResponse);
}

message NotifyRequest {
  string tenant_id        = 1;
  string notification_id  = 2;   // caller idempotency key
  repeated string user_ids = 3;
  repeated DeliveryChannel channels = 4;  // empty = all enabled channels
  string subject_ref      = 5;   // opaque; platform never interprets
  string subject_type     = 6;   // opaque; platform never interprets
  string title            = 7;
  string body             = 8;
  map<string, ChannelAddresses> addresses = 9;   // userID → {byChannel: {channel: address}}
}

message ChannelAddresses {
  map<string, string> by_channel = 1;
}

message NotifyResponse {
  int32 delivered = 1;   // provider accepted at least one (user, channel) row
  int32 pending   = 2;   // stored but no active provider or no destination
  int32 failed    = 3;   // provider returned an error
}

Wire path: POST /elloloop.notify.v1.NotificationInternalService/Notify

curl -sX POST http://notify:8081/elloloop.notify.v1.NotificationInternalService/Notify \
  -H 'Content-Type: application/json' \
  -H "X-Notify-Internal-Token: $NOTIFY_INTERNAL_TOKEN" \
  -d '{
    "tenantId": "acme",
    "notificationId": "task-42-comment-7",
    "userIds": ["user-alice", "user-bob"],
    "channels": ["DELIVERY_CHANNEL_IN_APP", "DELIVERY_CHANNEL_EMAIL"],
    "subjectRef": "task:42",
    "subjectType": "task",
    "title": "New comment",
    "body": "Bob commented on Task 42",
    "addresses": {
      "user-alice": { "byChannel": { "email": "alice@example.com" } },
      "user-bob":   { "byChannel": { "email": "bob@example.com" } }
    }
  }'

Errors:

InvalidArgument — tenant_id, notification_id, or user_ids empty.
Unauthenticated — missing or mismatched X-Notify-Internal-Token.
Internal — store error (DB unreachable, schema migration not applied, etc.).

NotificationClientService

Client-facing surface for browsers and mobile apps. Speaks Connect HTTP/2. Auth is Authorization: Bearer <JWT>. Tenant is derived from the JWT — request bodies never carry a tenant field.

StreamEvents

rpc StreamEvents(StreamEventsRequest) returns (stream StreamEvent);

message StreamEventsRequest {
  DeviceType device_type = 1;
}

message StreamEvent {
  string session_id = 1;
  oneof event {
    NotificationEvent notification = 2;
    DataChangeEvent   data_change  = 3;
    HeartbeatEvent    heartbeat    = 4;
  }
}

message NotificationEvent { Notification notification = 1; }

message DataChangeEvent {
  string idempotency_key = 1;
  string subject_ref     = 2;
  string subject_type    = 3;
}

message HeartbeatEvent { int64 timestamp_ms = 1; }

message Notification {
  string id              = 1;   // store id; pass to AckNotification
  string notification_id = 2;   // producer's idempotency key
  string subject_ref     = 3;
  string subject_type    = 4;
  string title           = 5;
  string body            = 6;
  DeliveryChannel channel = 7;
  DeliveryStatus  status  = 8;
  int64 created_at_ms     = 9;
  int64 delivered_at_ms   = 10;
  int64 ack_at_ms         = 11;
  int64 read_at_ms        = 12;
}

Long-lived server-stream. The first event is always a HeartbeatEvent carrying the assigned session_id so the client can target AckDataChange back at this exact connection.

Errors:

Unauthenticated — missing / invalid / expired JWT.
Unimplemented — NOTIFY_LIVE_CONNECTIONS_ENABLED=false.
Canceled — client disconnect, server shutdown.

GetNotifications

rpc GetNotifications(GetNotificationsRequest) returns (GetNotificationsResponse);

message GetNotificationsRequest {
  string cursor      = 1;   // opaque; round-trip the previous response's next_cursor
  int32  limit       = 2;   // 1..100; 0 → 20 default
  bool   unread_only = 3;
}

message GetNotificationsResponse {
  repeated Notification notifications = 1;
  string next_cursor = 2;   // empty on the last page
  int32  unread_count = 3;   // total not-yet-read, independent of page window
}

Newest-first. Cursor is strict < on created_at_ms. unread_count is the global not-yet-read total, not the count on this page.

curl -sX POST http://notify:8080/elloloop.notify.v1.NotificationClientService/GetNotifications \
  -H "Authorization: Bearer $JWT" \
  -H 'Content-Type: application/json' \
  -d '{"limit": 20}'

AckNotification

rpc AckNotification(AckNotificationRequest) returns (AckNotificationResponse);

message AckNotificationRequest {
  string id = 1;   // store id (Notification.id), NOT the producer's notification_id
}

message AckNotificationResponse {}

Marks one row as Read and stamps read_at_ms. Idempotent — repeated calls are no-ops. Cross-user / cross-tenant attempts return NotFound (never PermissionDenied, which would leak the existence of someone else's row).

AckDataChange

rpc AckDataChange(AckDataChangeRequest) returns (AckDataChangeResponse);

message AckDataChangeRequest {
  string idempotency_key = 1;
  string session_id      = 2;   // from StreamEvent.session_id
}

message AckDataChangeResponse {}

Cancels any pending retries for the (idempotency_key, session_id) pair on the in-memory RetryTracker. Sessions older than the process lifetime are silently ignored (the tracker treats unknown entries as no-ops).

RegisterPushToken

rpc RegisterPushToken(RegisterPushTokenRequest) returns (RegisterPushTokenResponse);

message RegisterPushTokenRequest {
  DeviceType device_type = 1;
  string token = 2;
}

message RegisterPushTokenResponse {}

Upserts the (user, device_type) → token row. Re-calling rotates the token in place. For Web Push the token is the JSON-stringified PushSubscription; for FCM it is the FCM registration token.

Error codes

Code	When
`InvalidArgument`	Missing required field, malformed cursor, empty user_ids on Notify, empty id on AckNotification.
`Unauthenticated`	Missing or invalid `Authorization` / `X-Notify-Internal-Token` header. Includes expired JWT.
`NotFound`	Acking a row that doesn't belong to the calling user (or doesn't exist).
`Unimplemented`	`StreamEvents` when `NOTIFY_LIVE_CONNECTIONS_ENABLED=false`.
`Canceled`	Stream interrupted (client close, server shutdown).
`Internal`	Store error, provider error not absorbed by the orchestrator, panic recovery.

Generating clients

Use buf against the published proto bundle on a tagged release:

# Download the proto bundle for the release you target.
gh release download v0.1.0 --repo elloloop/notify --pattern 'notify-protos-*.tar.gz'
tar xzf notify-protos-0.1.0.tar.gz

# Generate Go, TypeScript, Python stubs.
buf generate

Or import the generated stubs from the Go module:

import (
    notifyv1 "github.com/elloloop/notify/gen/go/notify/v1"
    "github.com/elloloop/notify/gen/go/notify/v1/notifyv1connect"
)