Documentation
¶
Index ¶
- Constants
- Variables
- func ArchivedActorNames() []string
- func AttentionNames() []string
- func CanonicalFields(payload []byte) (id, kind, state string)
- func FeedFields(payload []byte) (title, url string, updatedAt int64)
- func LaunchActionID(target string) string
- func LaunchActionTarget(actionID string) (string, bool)
- func LifecycleNames() []string
- func NotifyActionID(target string) string
- func NotifyActionTarget(actionID string) (string, bool)
- func ResurfacePolicyNames() []string
- func TransitionNames() []string
- type AbsenceConfirmer
- type AbsenceVerdict
- type ArchivedActor
- type Attention
- type Classification
- type Classifier
- type CommitBatch
- type Discard
- type FeedClaim
- type FeedSnapshot
- type ItemRef
- type KVMutation
- type Lifecycle
- type Msg
- type NodeRun
- type NotifyCommand
- type Observation
- type Output
- type ResurfacePolicy
- type Sink
- type SnapshotItem
- type Transition
Constants ¶
const ( SinkKindFeed = "feed" SinkKindAction = "action" SinkKindNotify = "notify" SinkKindLaunch = "launch" )
Sink kinds: where a committed Output is written.
const LaunchActionPrefix = "launch:"
LaunchActionPrefix is NotifyActionPrefix's counterpart for launch nodes.
const NotifyActionPrefix = "notify:"
NotifyActionPrefix namespaces the synthetic action id a notify terminal enqueues its output_command under, so one queue serves both authored actions.yml actions and inline notify nodes. Authored action ids are slugs (`^[a-z0-9][a-z0-9-]*$`), which can contain neither ":" nor "/", so a synthetic id can never collide with one.
Variables ¶
var ErrInvalidArchivedActor = fmt.Errorf("not a valid ArchivedActor, try [%s]", strings.Join(_ArchivedActorNames, ", "))
var ErrInvalidAttention = fmt.Errorf("not a valid Attention, try [%s]", strings.Join(_AttentionNames, ", "))
var ErrInvalidLifecycle = fmt.Errorf("not a valid Lifecycle, try [%s]", strings.Join(_LifecycleNames, ", "))
var ErrInvalidResurfacePolicy = fmt.Errorf("not a valid ResurfacePolicy, try [%s]", strings.Join(_ResurfacePolicyNames, ", "))
var ErrInvalidTransition = fmt.Errorf("not a valid Transition, try [%s]", strings.Join(_TransitionNames, ", "))
Functions ¶
func ArchivedActorNames ¶
func ArchivedActorNames() []string
ArchivedActorNames returns a list of possible string values of ArchivedActor.
func AttentionNames ¶
func AttentionNames() []string
AttentionNames returns a list of possible string values of Attention.
func CanonicalFields ¶
CanonicalFields is the one shared decode for the canonical top-level id/kind/state strings of an item payload (docs/decisions/0008). Every reader of the canonical contract goes through it — the action-item decode in dispatch and the webhook classifier in the webhook connector — so the string-or-number id rule has exactly one implementation.
A payload that is not a JSON object, or that omits a field or gives it a non-string value, yields "" for that field rather than an error: the canonical fields are optional by contract.
func FeedFields ¶
FeedFields is the one shared decode for the presentation an item payload carries: the title and url a feed row renders, and the time the item's own source says it last changed.
Both boundaries that mint an inbox row read it — the poll producer, for a message a source emitted, and CommitBatch, for a key a function node synthesized (ADR function-node-per-entity-feed-items) — so `updatedAt` on a per-entity payload means what it means on a source's own.
updatedAt is unix milliseconds. Zero means the payload did not say, and the caller substitutes its own clock rather than stamping the item at the epoch.
func LaunchActionID ¶
LaunchActionID takes a flow-qualified "<flowId>/<nodeId>" target.
func LaunchActionTarget ¶
func LifecycleNames ¶
func LifecycleNames() []string
LifecycleNames returns a list of possible string values of Lifecycle.
func NotifyActionID ¶
NotifyActionID returns the synthetic action id for a notify node, whose target is the flow-qualified node id "<flowId>/<nodeId>".
func NotifyActionTarget ¶
NotifyActionTarget inverts NotifyActionID: it returns the flow-qualified node id an action id names, and whether the id is a notify action at all.
func ResurfacePolicyNames ¶
func ResurfacePolicyNames() []string
ResurfacePolicyNames returns a list of possible string values of ResurfacePolicy.
func TransitionNames ¶
func TransitionNames() []string
TransitionNames returns a list of possible string values of Transition.
Types ¶
type AbsenceConfirmer ¶
type AbsenceConfirmer interface {
ConfirmAbsence(ctx context.Context, previous []Observation) (map[string]AbsenceVerdict, error)
}
Verdicts are keyed by Observation.ExternalID; an item the connector could not answer for is absent from the map. A returned error is a partial failure — the verdicts already in the map are valid.
type AbsenceVerdict ¶
type AbsenceVerdict struct {
Current *Observation
Terminal bool
}
type ArchivedActor ¶
type ArchivedActor string
ENUM(manual, system)
const ( // ArchivedActorManual is a ArchivedActor of type manual. ArchivedActorManual ArchivedActor = "manual" // ArchivedActorSystem is a ArchivedActor of type system. ArchivedActorSystem ArchivedActor = "system" )
func ParseArchivedActor ¶
func ParseArchivedActor(name string) (ArchivedActor, error)
ParseArchivedActor attempts to convert a string to a ArchivedActor.
func (*ArchivedActor) AppendText ¶
func (x *ArchivedActor) AppendText(b []byte) ([]byte, error)
AppendText appends the textual representation of itself to the end of b (allocating a larger slice if necessary) and returns the updated slice.
Implementations must not retain b, nor mutate any bytes within b[:len(b)].
func (ArchivedActor) IsValid ¶
func (x ArchivedActor) IsValid() bool
IsValid provides a quick way to determine if the typed value is part of the allowed enumerated values
func (ArchivedActor) MarshalText ¶
func (x ArchivedActor) MarshalText() ([]byte, error)
MarshalText implements the text marshaller method.
func (ArchivedActor) String ¶
func (x ArchivedActor) String() string
String implements the Stringer interface.
func (*ArchivedActor) UnmarshalText ¶
func (x *ArchivedActor) UnmarshalText(text []byte) error
UnmarshalText implements the text unmarshaller method.
type Attention ¶
type Attention string
ENUM(activity, trivial)
func ParseAttention ¶
ParseAttention attempts to convert a string to a Attention.
func (*Attention) AppendText ¶
AppendText appends the textual representation of itself to the end of b (allocating a larger slice if necessary) and returns the updated slice.
Implementations must not retain b, nor mutate any bytes within b[:len(b)].
func (Attention) IsValid ¶
IsValid provides a quick way to determine if the typed value is part of the allowed enumerated values
func (Attention) MarshalText ¶
MarshalText implements the text marshaller method.
func (*Attention) UnmarshalText ¶
UnmarshalText implements the text unmarshaller method.
type Classification ¶
type Classification struct {
Kind string
Transition Transition
Attention Attention
Lifecycle Lifecycle
SourceState string
ArchivedReason string
OccurrenceKey string
Summary string
Detail []byte
}
Classification is the persisted result of comparing normalized source state.
type Classifier ¶
type Classifier interface {
Classify(previous *Observation, current Observation) Classification
}
Classifier is deliberately independent of source wire formats.
type CommitBatch ¶
type CommitBatch struct {
Consumer string `json:"consumer"` // event_log consumer key (flow id / consumer id)
UpToOffset int64 `json:"upToOffset"`
Outputs []Output `json:"outputs"`
FeedSnapshots []FeedSnapshot `json:"feedSnapshots"`
Discards []Discard `json:"discards"`
NodeRuns []NodeRun `json:"nodeRuns"`
KVMutations []KVMutation `json:"kvMutations,omitempty"` // omitempty: invisible to fixtures with none
}
CommitBatch is the graph runtime's (cmd/desktop/internal/app/runtime) atomic write: it advances a consumer's committed offset and persists the outputs/node-run metrics produced while processing up to that offset, all in one transaction (see stores.EventLogStore.Commit).
type Discard ¶
Discard records a message a node dropped instead of forwarding, for metrics/observability. CommitBatch does not persist Discards as rows — they exist for callers that want to log or count them; the per-node aggregate is expected to already be reflected in the corresponding NodeRun.DropCount.
type FeedClaim ¶
type FeedClaim struct {
ProfileID string `json:"profileId"`
FeedID string `json:"feedId"`
ItemID int64 `json:"itemId"`
SourceID string `json:"sourceId"`
}
FeedClaim is one item's membership in one feed, attributed to the source that produced it. The engine builds these for ActivateReplay; no store reads one back.
type FeedSnapshot ¶
type FeedSnapshot struct {
FeedID string `json:"feedId"`
SourceTopic string `json:"sourceTopic"`
SnapshotID string `json:"snapshotId"`
}
FeedSnapshot declares one source's complete current output scope for a feed. EventLogStore.Commit reconciles the (feed, source) scope's membership claims down to the feed outputs carrying this SnapshotID.
type ItemRef ¶
type ItemRef struct {
ProfileID string `json:"profileId"`
SourceKind string `json:"sourceKind"`
SourceScope string `json:"sourceScope"`
ExternalID string `json:"externalId"`
}
ItemRef identifies an inbox item by inbox_item's own UNIQUE key rather than by its row id, which ActivateReplay does not preserve (ADR an-item-session-link-is-desktop-state-keyed-on-item-coordinates).
type KVMutation ¶
type KVMutation struct {
NodeID string `json:"nodeId"`
Key string `json:"key"`
Delete bool `json:"delete,omitempty"`
Value string `json:"value,omitempty"`
ExpiresAt int64 `json:"expiresAt,omitempty"` // unix ms, 0 = no expiry
}
KVMutation is one node's durable KV write, flushed inside CommitBatch so it is durable iff the tick that produced it commits. The row's flow_id is CommitBatch.Consumer.
type Lifecycle ¶
type Lifecycle string
ENUM(active, terminal, unknown)
func ParseLifecycle ¶
ParseLifecycle attempts to convert a string to a Lifecycle.
func (*Lifecycle) AppendText ¶
AppendText appends the textual representation of itself to the end of b (allocating a larger slice if necessary) and returns the updated slice.
Implementations must not retain b, nor mutate any bytes within b[:len(b)].
func (Lifecycle) IsValid ¶
IsValid provides a quick way to determine if the typed value is part of the allowed enumerated values
func (Lifecycle) MarshalText ¶
MarshalText implements the text marshaller method.
func (*Lifecycle) UnmarshalText ¶
UnmarshalText implements the text unmarshaller method.
type Msg ¶
type Msg struct {
ID string
Key string
Topic string
Ts int64
Payload json.RawMessage
// Snapshot must NOT be omitempty: an empty snapshot (a successful poll
// that returned zero items) marshals as [] and the frontend engine's
// `msg.Snapshot != null` routing depends on it. With omitempty the field
// vanishes, the boundary row (key "") is routed as an ordinary item, and
// CommitBatch fails resolving inbox item "<kind>//" forever — wedging the
// consumer at that offset.
Snapshot []SnapshotItem
SourceKind string
SourceScope string
OccurrenceKey string `json:"OccurrenceKey,omitempty"`
}
Msg is the pipeline's generic log record, appended by sources and consumed by the graph runtime (cmd/desktop/internal/app/runtime).
It mirrors the design's { id, key, topic, ts, payload } contract, with Snapshot populated only for an authoritative full-source snapshot, mapped onto the event_log schema (see queries/migrations/0001_pipeline.up.sql):
- ID is derived from the row's "offset" (stable, unique per append; there is no separate id column) and stays a decimal string rather than an int64. A function node's script sees this value as a genuine JavaScript value crossing goja's JSON.parse/JSON.stringify boundary (runtime/js.go), including round-tripping through a script that returns the message it was handed — and a JS number cannot represent an int64 exactly past 2^53. The string is what keeps a large offset intact for the script.
- Ts is the row's created_at (unix milliseconds).
- Snapshot is nil for ordinary item events and contains the full current source item set for successful poll snapshots.
type NodeRun ¶
type NodeRun struct {
FlowID string `json:"flowId"`
NodeID string `json:"nodeId"`
OK bool `json:"ok"`
InCount int `json:"inCount"`
OutCount int `json:"outCount"`
DropCount int `json:"dropCount"`
Err string `json:"err"`
DurMs int64 `json:"durMs"`
}
NodeRun is one node's per-tick execution summary, recorded for the flows debug/status UI.
type NotifyCommand ¶
type NotifyCommand struct {
ProfileID string `json:"profileId"`
ExternalID string `json:"externalId,omitempty"`
SourceKind string `json:"sourceKind,omitempty"`
SourceScope string `json:"sourceScope,omitempty"`
Item json.RawMessage `json:"item,omitempty"`
}
NotifyCommand is the durable payload of a notify terminal's output_command. It carries the triggering item's identity alongside its payload: the identity resolves the inbox row a delivered notification links back to (so clicking the banner can select that item), while Item is what the node's title/body templates render over — the same shape an action node's executor sees, so `{{ .Payload.title }}` means the same thing in both.
type Observation ¶
type Observation struct {
ExternalID string
Title string
URL string
SourceKind string
SourceScope string
ObservedAt int64
Payload []byte
}
Observation is an adapter-neutral representation of a source item.
type Output ¶
type Output struct {
Sink Sink `json:"sink"`
Key string `json:"key,omitempty"` // the item's external ID; not set by a keyless snapshot boundary
OccurrenceKey string `json:"occurrenceKey,omitempty"` // action dedup key only
Payload json.RawMessage `json:"payload,omitempty"` // action payload only
SourceKind string `json:"sourceKind,omitempty"`
SourceScope string `json:"sourceScope,omitempty"`
SourceTopic string `json:"sourceTopic"`
SnapshotID string `json:"snapshotId,omitempty"`
}
Output is one committed side effect of a flow run.
type ResurfacePolicy ¶
type ResurfacePolicy string
ENUM(all, state-changes, never)
const ( // ResurfacePolicyAll is a ResurfacePolicy of type all. ResurfacePolicyAll ResurfacePolicy = "all" // ResurfacePolicyStateChanges is a ResurfacePolicy of type state-changes. ResurfacePolicyStateChanges ResurfacePolicy = "state-changes" // ResurfacePolicyNever is a ResurfacePolicy of type never. ResurfacePolicyNever ResurfacePolicy = "never" )
func ParseResurfacePolicy ¶
func ParseResurfacePolicy(name string) (ResurfacePolicy, error)
ParseResurfacePolicy attempts to convert a string to a ResurfacePolicy.
func (*ResurfacePolicy) AppendText ¶
func (x *ResurfacePolicy) AppendText(b []byte) ([]byte, error)
AppendText appends the textual representation of itself to the end of b (allocating a larger slice if necessary) and returns the updated slice.
Implementations must not retain b, nor mutate any bytes within b[:len(b)].
func (ResurfacePolicy) IsValid ¶
func (x ResurfacePolicy) IsValid() bool
IsValid provides a quick way to determine if the typed value is part of the allowed enumerated values
func (ResurfacePolicy) MarshalText ¶
func (x ResurfacePolicy) MarshalText() ([]byte, error)
MarshalText implements the text marshaller method.
func (ResurfacePolicy) String ¶
func (x ResurfacePolicy) String() string
String implements the Stringer interface.
func (*ResurfacePolicy) UnmarshalText ¶
func (x *ResurfacePolicy) UnmarshalText(text []byte) error
UnmarshalText implements the text unmarshaller method.
type Sink ¶
Sink identifies where an Output is committed. Feed outputs claim immutable inbox membership; action, notify, and launch outputs enqueue an output_command.
type SnapshotItem ¶
type SnapshotItem struct {
Key string `json:"key"`
Payload json.RawMessage `json:"payload"`
}
SnapshotItem is one current source item carried by a successful source snapshot. A snapshot event is distinct from ordinary changed-item events: it is emitted on every successful poll, including when the source is empty.
type Transition ¶
type Transition string
ENUM(none, entered-terminal, left-terminal)
const ( // TransitionNone is a Transition of type none. TransitionNone Transition = "none" // TransitionEnteredTerminal is a Transition of type entered-terminal. TransitionEnteredTerminal Transition = "entered-terminal" // TransitionLeftTerminal is a Transition of type left-terminal. TransitionLeftTerminal Transition = "left-terminal" )
func ParseTransition ¶
func ParseTransition(name string) (Transition, error)
ParseTransition attempts to convert a string to a Transition.
func (*Transition) AppendText ¶
func (x *Transition) AppendText(b []byte) ([]byte, error)
AppendText appends the textual representation of itself to the end of b (allocating a larger slice if necessary) and returns the updated slice.
Implementations must not retain b, nor mutate any bytes within b[:len(b)].
func (Transition) IsValid ¶
func (x Transition) IsValid() bool
IsValid provides a quick way to determine if the typed value is part of the allowed enumerated values
func (Transition) MarshalText ¶
func (x Transition) MarshalText() ([]byte, error)
MarshalText implements the text marshaller method.
func (Transition) String ¶
func (x Transition) String() string
String implements the Stringer interface.
func (*Transition) UnmarshalText ¶
func (x *Transition) UnmarshalText(text []byte) error
UnmarshalText implements the text unmarshaller method.