Documentation
¶
Overview ¶
Package exportstream is what every Export stream shares (ADR 0045, GLOSSARY: Export stream): the opaque (updated_at, id) cursor, the keyset query predicates, the page size, the NDJSON page encoder and the check of an encoded row against the TypeSpec row schema. A stream adds only its row type (with Position), its fetch query and its operation; see events.go for the smallest example.
Core code: the license gate is the caller's (internal/api/external asks the Edition), nothing here knows about licenses.
Index ¶
- Constants
- Variables
- func Order(s *sql.Selector)
- func Seek(after Cursor) func(*sql.Selector)
- func ValidateRow(schema string, line []byte) error
- type AuditRow
- type ContactRow
- type Cursor
- type ErasureRow
- type EventRow
- type MessageRow
- type Page
- func Audit(ctx context.Context, s *ent.Scoped, token string, limit int) (Page, error)
- func Contacts(ctx context.Context, s *ent.Scoped, token string, limit int) (Page, error)
- func Erasures(ctx context.Context, s *ent.Scoped, token string, limit int) (Page, error)
- func Events(ctx context.Context, s *ent.Scoped, token string, limit int) (Page, error)
- func Messages(ctx context.Context, s *ent.Scoped, token string, limit int) (Page, error)
- func Read[R Row](token string, limit int, fetch func(after Cursor, limit int) ([]R, error)) (Page, error)
- type Row
Constants ¶
const AuditSchema = "ExportAuditRow"
AuditSchema is the TypeSpec row schema (openapi component) an AuditRow encodes to.
const ContactsSchema = "ExportContactRow"
ContactsSchema is the TypeSpec row schema (openapi component) a ContactRow encodes to.
const ErasuresSchema = "ExportErasureRow"
ErasuresSchema is the TypeSpec row schema (openapi component) an ErasureRow encodes to.
const EventsSchema = "ExportEventRow"
EventsSchema is the TypeSpec row schema (openapi component) an EventRow encodes to.
const MessagesSchema = "ExportMessageRow"
MessagesSchema is the TypeSpec row schema (openapi component) a MessageRow encodes to.
Variables ¶
var ErrBadCursor = errors.New("exportstream: invalid cursor")
ErrBadCursor means a cursor token is not one this package issued: a malformed or foreign value is refused (400) instead of silently restarting a consumer from zero.
Functions ¶
func Seek ¶
Seek is the keyset predicate of a stream's query: rows at or after the cursor. Pass it to an ent query's Where as a predicate of that entity (`predicate.Event(exportstream.Seek(c))`); it adds nothing for the beginning of the stream. It reads the entity's `updated_at` and `id` columns, which the index on (workspace_id, updated_at, id) covers.
func ValidateRow ¶
ValidateRow reports why one encoded row (a line of a page, without its newline) does not match the TypeSpec row schema named by schema (an openapi component such as EventsSchema), or nil when it does. It is the check that keeps a stream's row type and its published contract the same thing: beyond the schema's own types and required keys, a key the schema does not declare is a failure too, because the generated documentation would not describe it.
Types ¶
type AuditRow ¶
type AuditRow struct {
ID int64 `json:"id,string"`
OccurredAt time.Time `json:"occurred_at"`
ActorKind string `json:"actor_kind"`
ActorID *string `json:"actor_id"`
ActorName *string `json:"actor_name"`
Action string `json:"action"`
TargetType string `json:"target_type"`
TargetID *string `json:"target_id"`
TargetName *string `json:"target_name"`
Diff map[string]json.RawMessage `json:"diff"`
RequestID *string `json:"request_id"`
IP *string `json:"ip"`
UserAgent *string `json:"user_agent"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
AuditRow is one line of the Audit stream, the contract's ExportAuditRow: every key is always present, an absent value is null, ids are decimal strings like everywhere in /api. Entries are append-only, so a row is delivered once.
type ContactRow ¶
type ContactRow struct {
ID int64 `json:"id,string"`
SubjectID *string `json:"subject_id"`
Email *string `json:"email"`
Phone *string `json:"phone"`
FirstName *string `json:"first_name"`
LastName *string `json:"last_name"`
TimeZone *string `json:"time_zone"`
CustomFields map[string]any `json:"custom_fields"`
Tags []string `json:"tags"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
ContactRow is one line of the Contacts stream, the contract's ExportContactRow: every key is always present, an absent value is null, Custom fields are one JSON column and the Tags are their names.
func (ContactRow) Position ¶
func (r ContactRow) Position() Cursor
Position is the row's (updated_at, id).
type Cursor ¶
Cursor is a position in a stream: every row at or after (At, ID) in (updated_at, id) order. The zero Cursor is the beginning of the stream. It is inclusive, and the cursor a page hands back is the position right after its last row (Following), so a row that shares its updated_at with the last row of a page is never skipped and never repeated.
func ParseCursor ¶
ParseCursor reads a token String issued. An empty token is the beginning of the stream; anything else that String could not have produced is ErrBadCursor.
func (Cursor) Following ¶
Following is the position right after c: ids are integers, so the successor of (updated_at, id) among the rows of one instant is (updated_at, id+1). The instant is cut to microseconds, the precision Postgres stores.
type ErasureRow ¶
type ErasureRow struct {
ID int64 `json:"id,string"`
ContactID *int64 `json:"contact_id,string"`
SubjectID string `json:"subject_id"`
VisitorIDs []string `json:"visitor_ids"`
EmailHash *string `json:"email_hash"`
PhoneHash *string `json:"phone_hash"`
ErasedAt time.Time `json:"erased_at"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
ErasureRow is one line of the Erasures stream, the contract's ExportErasureRow: the keys an Erasure matched on. Every key is always present, an absent value is null, ids are decimal strings like everywhere in /api. The hashes are SHA-256 as lowercase hex; the address and the phone themselves are never stored, so they cannot be here.
func (ErasureRow) Position ¶
func (r ErasureRow) Position() Cursor
Position is the row's (updated_at, id).
type EventRow ¶
type EventRow struct {
ID int64 `json:"id,string"`
Action string `json:"action"`
Properties map[string]any `json:"properties"`
ContactID *int64 `json:"contact_id,string"`
VisitorID *string `json:"visitor_id"`
SubjectID string `json:"subject_id"`
Email *string `json:"email"`
Phone *string `json:"phone"`
OccurredAt *time.Time `json:"occurred_at"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
EventRow is one line of the Events stream, the contract's ExportEventRow: every key is always present, an absent value is null, ids are decimal strings like everywhere in /api.
type MessageRow ¶
type MessageRow struct {
ID int64 `json:"id,string"`
Kind string `json:"kind"`
Channel string `json:"channel"`
Destination *string `json:"destination"`
Status string `json:"status"`
Reason *string `json:"reason"`
ContactID *int64 `json:"contact_id,string"`
BroadcastID *int64 `json:"broadcast_id,string"`
AutomationID *int64 `json:"automation_id,string"`
AutomationRunID *int64 `json:"automation_run_id,string"`
AutomationStep *int `json:"automation_step"`
IntegrationID *int64 `json:"integration_id,string"`
TemplateID *int64 `json:"template_id,string"`
SendingSource *string `json:"sending_source"`
SendingDomain *string `json:"sending_domain"`
ProviderMessageID *string `json:"provider_message_id"`
ClaimedAt time.Time `json:"claimed_at"`
SentAt *time.Time `json:"sent_at"`
DeliveredAt *time.Time `json:"delivered_at"`
BouncedAt *time.Time `json:"bounced_at"`
ComplainedAt *time.Time `json:"complained_at"`
Captured bool `json:"captured"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
MessageRow is one line of the Outbound messages stream, the contract's ExportMessageRow: delivery provenance and outcome, never a subject or a body. Every key is always present, an absent value is null, ids are decimal strings like everywhere in /api.
func (MessageRow) Position ¶
func (r MessageRow) Position() Cursor
Position is the row's (updated_at, id).
type Page ¶
Page is one answer of a stream: the body (NDJSON, one row per line, no envelope) and the token of the cursor for the next request.
func Audit ¶
Audit serves a page of the Workspace's Audit stream. The caller has checked the `audit` license, which is an Enterprise feature of the entries themselves.
func Contacts ¶
Contacts serves a page of the Workspace's Contacts stream. A Tag change moves the Contact's updated_at (internal/tags), so the Tag names ride the row's own version.
func Read ¶
func Read[R Row](token string, limit int, fetch func(after Cursor, limit int) ([]R, error)) (Page, error)
Read serves one page of a stream. token is the request's cursor; fetch returns the rows at or after a cursor in (updated_at, id) order, at most limit of them. The page's cursor is the position after the last row, or token itself when the page is empty, so a poller can loop on it without special cases. A malformed token is ErrBadCursor.
Source Files