Documentation
¶
Overview ¶
Package notification delivers in-app messages to users: the outcome of a sync operation somebody asked for, and the failures a workspace administrator needs to see. It is the one writer of the notification table, so the recipient resolution, the suppression of repeated background messages and the envelope they share live here rather than in each caller.
Delivery is best effort by contract. A caller is on a sync, an ingestion or a cleanup path, and none of them may fail because a message could not be written: they log the error and carry on, the same way the audit interceptor treats its ledger writes.
A written message is also handed to the recipient's live subscriptions (see Subscribe), which is how a signed-in client learns about it without polling. That channel is process-local and cannot fail a write: a subscription that has stopped reading is dropped rather than allowed to hold the writer open.
Index ¶
- Constants
- func DedupeKey(event, target string, now time.Time, window time.Duration) string
- func LogFailure(err error, attrs ...any)
- func OpenLineageMessage(severity storepb.NotificationSeverity, detail *storepb.OpenLineageDetail) *storepb.Notification
- func SchemaSyncMessage(recipientID int, severity storepb.NotificationSeverity, ...) *storepb.Notification
- type AdminStore
- type Event
- type Service
- func (s *Service) Close()
- func (s *Service) ReportUnmatchedNamespace(ctx context.Context, namespace, dataset string)
- func (s *Service) Send(ctx context.Context, n *storepb.Notification) error
- func (s *Service) SendToWorkspaceAdmins(ctx context.Context, n *storepb.Notification) error
- func (s *Service) Subscribe(recipientID int) (<-chan Event, func(), error)
- func (s *Service) WorkspaceAdminIDs(ctx context.Context) ([]int, error)
Constants ¶
const ( // BackgroundFailureWindow suppresses a repeated failure report of the same // target. A failing instance is retried by the periodic scan every fifteen // minutes and a failing database backs off for a few, so without a window a // single broken target would report every few minutes forever. BackgroundFailureWindow = time.Hour // UnmappedNamespaceWindow is longer than the failure window on purpose: a // dataset that matched no instance is a configuration mistake rather than an // outage, and it repeats for every event the producer sends. UnmappedNamespaceWindow = 24 * time.Hour // MaxFailureEntries bounds how many databases one sync message lists. // The message's counts carry the rest, so it stays a pointer into the estate // rather than a copy of it. MaxFailureEntries = 100 // MaxErrorBytes bounds one recorded error. A driver error can quote a whole // statement, and the row is kept forever, so it is trimmed at the door. MaxErrorBytes = 2 << 10 )
const ( // MaxSubscriptionsPerRecipient bounds what one account may hold open. One // browser tab holds one, and a reconnect overlaps briefly with the stream it // replaces. Past the bound the newest connection is refused rather than the // oldest one dropped: the caller is told, instead of another tab going quiet. MaxSubscriptionsPerRecipient = 8 )
Variables ¶
This section is empty.
Functions ¶
func DedupeKey ¶
DedupeKey builds the suppression key of a background notification: <event>:<target>:<bucket>, where the bucket is the start of the window the occurrence falls in. The unique index on (recipient_id, dedupe_key) turns a second occurrence inside one window into a no-op, so a broken producer or a failing instance cannot fill an inbox, and it does so across replicas because the suppression is the index rather than process memory.
func LogFailure ¶
LogFailure records a message that could not be delivered. Callers on a sync or ingestion path use it instead of propagating the error.
func OpenLineageMessage ¶
func OpenLineageMessage(severity storepb.NotificationSeverity, detail *storepb.OpenLineageDetail) *storepb.Notification
OpenLineageMessage builds the message an OpenLineage ingestion failure is reported with. It is always addressed to the workspace administrators: an ingestion key is a machine credential, so no user asked for the request.
func SchemaSyncMessage ¶
func SchemaSyncMessage(recipientID int, severity storepb.NotificationSeverity, detail *storepb.SchemaSyncDetail) *storepb.Notification
SchemaSyncMessage builds the message one sync operation reports its outcome with. recipientID is 0 when the message is addressed to the workspace administrators rather than to the user who asked for the sync.
Types ¶
type AdminStore ¶
type AdminStore interface {
CreateNotification(ctx context.Context, notification *storepb.Notification) (*storepb.Notification, error)
CountNotificationCounts(ctx context.Context, recipientID int) (unread int, unseen int, err error)
GetWorkspaceID(ctx context.Context) (string, error)
GetWorkspaceIamPolicy(ctx context.Context) (*store.IamPolicyMessage, error)
GetGroup(ctx context.Context, email string) (*store.GroupMessage, error)
GetUserByID(ctx context.Context, id int) (*store.UserMessage, error)
}
AdminStore is the subset of *store.Store this package reads. Narrowing it keeps the recipient resolution testable without a database and documents exactly what delivering a message costs.
type Event ¶
type Event struct {
Notification *storepb.Notification
// UnreadCount is every unread message, and UnseenCount is the subset written
// since the recipient last opened the inbox. They differ once the recipient
// has opened the inbox and left a message unread: nothing is marked read there.
UnreadCount int32
UnseenCount int32
// CountsKnown is false when the counts could not be read. The message itself is
// still worth delivering; the badge is not, because a zero would read as
// "everything is read".
CountsKnown bool
}
Event is one message handed to the live subscriptions of an inbox: the row a write produced, and the recipient's counts right after it.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service writes notifications and works out who receives them. It is also where the live push channel starts: a message it writes is handed to the subscriptions of its recipient, so a signed-in client sees it without asking.
func (*Service) Close ¶
func (s *Service) Close()
Close ends every live subscription. It is what lets the server drain: the HTTP server's shutdown waits for in-flight requests, and a notification stream is one that never ends on its own.
func (*Service) ReportUnmatchedNamespace ¶
ReportUnmatchedNamespace tells the workspace administrators that an ingested dataset matched no registered instance and was stored as an external dataset. It is the OpenLineage resolver's reporter: the resolver decides when a namespace is worth reporting, this decides how.
func (*Service) Send ¶
Send delivers one notification to one recipient. A notification whose dedupe key cannot be claimed is silently dropped: the message it repeats is already in the inbox.
func (*Service) SendToWorkspaceAdmins ¶
SendToWorkspaceAdmins delivers a copy of the notification to every workspace administrator. The message may arrive with recipient_id unset: it is stamped per recipient here.
func (*Service) Subscribe ¶
Subscribe registers a live subscription for one recipient's inbox. The returned channel is closed when the subscription ends — the server is shutting down, the connection could not keep up, or the caller unsubscribed — and the returned function releases it. Releasing is safe to call more than once.
func (*Service) WorkspaceAdminIDs ¶
WorkspaceAdminIDs resolves the principals holding roles/workspaceAdmin in the workspace IAM policy, expanding group members. The allUsers pseudo-member cannot be bound to that role, and the system bot cannot sign in, so both are skipped rather than notified.
Source Files