Help us shape the future of Go! Take the 2026 Go Developer Survey

emailrules

package
v0.28.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 14, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ModeClusterRoles = "cluster_roles" // operational lane: the transactional outbox
	ModeAudience     = "audience"      // marketing lane: the campaign machinery
	ModeRowAddress   = "row_address"   // marketing lane, address off the triggering row
)

The three recipient modes, and the two lanes they pick (program P5). The lane is a CONSEQUENCE of who receives, never a separate setting -- a person choosing "the people in this cluster" has chosen the operational lane whether or not they have ever heard the word.

View Source 
const BreakerStoppedSentence = "Stopped after repeated failures. Fix the cause, then turn it on again."

BreakerStoppedSentence is what a rule a breaker stopped says on its row: what happened and what to do.

View Source 
const IntegrationName = "emailrules"

IntegrationName is the DSL-visible namespace: integration.emailrules.*

View Source 
const PreviousVersionStillRunning = "The previous version is still running."

PreviousVersionStillRunning ends the reason a refused edit records: the version that was live before the edit is still armed and still sending.

View Source 
const RuleConcept = "v1:campaigns:emailRule"

RuleConcept is the email rule's concept id: what a write conditional on the rule row is fenced on.

Variables

This section is empty.

Functions

func Bind 

func Bind(engine ActivationEngine, deps func() memql.AuthoredRuntimeDeps)

Bind hands this package the concrete engine and the authored-runtime deps. Called once from the App during the engine phase, right after the authored runtime exists. Idempotent; a second call replaces the first, which is what a test that stands two engines up needs.

The same handoff registers, on an engine that takes one, the listener that marks a rule paused when this node records its breaker's pause (breaker_pause.go). It is registered by name, so binding again replaces it.

func BundleTitleFor 

func BundleTitleFor(r Rule) string

BundleTitleFor is what an operator sees in the authoring surfaces. It names the rule rather than the construct, because the rule is the thing they made.

func ConstructNameFor 

func ConstructNameFor(ruleID string) string

ConstructNameFor derives the generated automation's name from the rule id.

DETERMINISTIC, because that is what makes a regeneration REPLACE rather than accumulate: the activation planner supersedes a prior bundle by matching what it registers, and two names for one rule would leave the old automation armed beside the new one -- a rule that had been edited would then send twice, once with each version, which is the failure mode a person reports as "it sent the old copy as well".

Construct names are unique tree-wide, so the prefix is not decoration.

func GenerateAutomation 

func GenerateAutomation(r Rule) (string, error)

GenerateAutomation renders the rule's executable form.

Pure: a function of the Rule and nothing else, so its whole output is diffable in a test and a change to it is visible as a change to the expected string rather than as a behaviour somebody has to reproduce.

func LaneFor 

func LaneFor(recipientMode string) string

LaneFor reports which delivery lane a recipient mode implies. Operational mail must never consult or write the marketing suppression list: an unsubscribe from a newsletter that silenced a security notice would be a correctly-implemented disaster.

func NormalizedRoles 

func NormalizedRoles(roles []string) []string

NormalizedRoles returns the operational lane's recipient roles, lowercased, de-duplicated and ordered, so two spellings of one rule produce one answer.

Types

type ActivationEngine 

type ActivationEngine interface {
	Engine
	ActivateApprovedBundle(ctx context.Context, owner, bundleId string, deps memql.AuthoredRuntimeDeps) (memql.ActivationResult, error)
	RetireActiveBundle(ctx context.Context, owner, bundleId string, deps memql.AuthoredRuntimeDeps) error
}

ActivationEngine is the concrete engine surface the activation path needs. Narrow on purpose: this package has no business with the rest of the engine, and a narrow seam is one a test can stand in for.

type Activator 

type Activator struct {
	// contains filtered or unexported fields
}

Activator arms and retires rules.

func NewActivator 

func NewActivator(engine ActivationEngine, deps func() memql.AuthoredRuntimeDeps) *Activator

func (*Activator) Activate 

func (a *Activator) Activate(ctx context.Context, owner, ruleID string) (Result, error)

Activate generates, validates, writes and arms a rule's construct.

`owner` is the AUTHENTICATED caller. It must equal the rule's owner, and the check is not decoration: the generated automation runs under the AUTHOR's envelope, so activating somebody else's rule would be running their reads under their identity on your say-so. `ActivateApprovedBundle` enforces the same rule at the bundle grain; this refuses earlier, with a sentence about rules rather than about bundles.

func (*Activator) Retire 

func (a *Activator) Retire(ctx context.Context, owner, ruleID string) (Result, error)

Retire tears a rule's construct down. The RULE row survives with its history: a rule somebody turned off is a different thing from one that never existed, and an app that deleted the row would have nothing to show for the mail it already sent.

It retires the bundle the rule row NAMES, never one derived again: a bundle keeps the id it was armed under, which for a rule armed before ids carried their owner is the old derivation (legacyBundleIDFor). The activation after it derives afresh, with the owner.

type ConditionError added in v0.22.0 

type ConditionError struct{ Message string }

ConditionError is a refused condition, worded for the person who typed it: the condition is the one place an end user writes an expression in the product (the OS Campaigns app's "Only when" field), and this message is what that field shows. Plain words, and the fix in one sentence.

func (*ConditionError) Error added in v0.22.0 

func (e *ConditionError) Error() string

type Engine 

type Engine interface {
	Execute(ctx context.Context, query string) (any, error)
}

Engine is the sliver of the engine this package needs.

type EngineAdapter 

type EngineAdapter struct {
	Engine *memql.MemQLEngine
}

EngineAdapter fits the live *memql.MemQLEngine to ActivationEngine.

The two differ in exactly one thing: the engine returns a *ExecuteResult where this package's read helpers take `any` -- and MaterializeRows accepts either. So the adapter is one method wide, and it lives here rather than in app/ for the same reason CampaignsEngineAdapter lives beside its consumer: the shape being adapted is this package's, and a wrapper written at the call site is one that has to be rewritten at the next call site.

func (EngineAdapter) ActivateApprovedBundle 

func (a EngineAdapter) ActivateApprovedBundle(ctx context.Context, owner, bundleId string, deps memql.AuthoredRuntimeDeps) (memql.ActivationResult, error)

func (EngineAdapter) Execute 

func (a EngineAdapter) Execute(ctx context.Context, query string) (any, error)

func (EngineAdapter) RetireActiveBundle 

func (a EngineAdapter) RetireActiveBundle(ctx context.Context, owner, bundleId string, deps memql.AuthoredRuntimeDeps) error

func (EngineAdapter) SetAuthoredPauseListener added in v0.26.0 

func (a EngineAdapter) SetAuthoredPauseListener(name string, listener memql.AuthoredPauseListener)

SetAuthoredPauseListener hands the live engine a breaker-pause listener.

type FireOutcome 

type FireOutcome struct {
	RuleID     string
	Lane       string
	Recipients int
	Sent       int
	Deferred   int
	Skipped    int
	Refusals   []string
	// Duplicate is a created rule's second firing for one row -- two writes
	// raced on a new row -- which sends nothing and is not counted as a firing.
	Duplicate bool
	// Busy: a send found the cluster's lock busy and, after a few short pauses,
	// still busy. Nothing was attempted for that recipient, the rule's fault it is
	// not, and a created rule's claim for the row was given back so the row can fire
	// again.
	Busy bool
}

FireOutcome is what one firing did, for the result node and the log line.

type Firer 

type Firer struct {
	// contains filtered or unexported fields
}

Firer executes one rule against one triggering event.

func NewFirer 

func NewFirer(engine Engine) *Firer

func (*Firer) Fire 

func (f *Firer) Fire(ctx context.Context, ruleID, nodeID string, event map[string]any) (FireOutcome, error)

Fire resolves the rule, decides the lane from who receives, and sends.

`nodeId` and `event` come from the generated automation's one step. `event` is the whole envelope; its `payload` is the triggering row as the trigger delivered it.

func (*Firer) WithFiringGate added in v0.23.6 

func (f *Firer) WithFiringGate(gate FiringGate) *Firer

WithFiringGate returns the Firer with the gate that records one rule's firings one at a time across every replica (firing_gate.go). Without one, a firing is recorded unlocked, and firings that overlap can lose increments.

func (*Firer) WithFirstFireClaim added in v0.22.0 

func (f *Firer) WithFirstFireClaim(claim FirstFireClaim) *Firer

WithFirstFireClaim returns the Firer with the claim a created rule needs.

func (*Firer) WithFirstFireRelease added in v0.26.0 

func (f *Firer) WithFirstFireRelease(release FirstFireRelease) *Firer

WithFirstFireRelease returns the Firer with the release that goes with the claim. Without one, a created rule's firing that was held up by a busy lock keeps its claim, and the row can never fire again.

type FiringGate added in v0.23.6 

type FiringGate func(ctx context.Context, ruleID string, record func(context.Context) error) error

FiringGate runs record while no other firing of the same rule -- on this node or any other -- is inside its own record.

type FiringState added in v0.23.6 

type FiringState struct {
	FiredCount int
	// VersionAt is the createdAt of the version read -- the version the
	// record must land after (firedAtAfter). Zero when the read did not carry
	// one, which degrades to stamping at the local clock.
	VersionAt time.Time
}

FiringState is what a firing's record is derived from: the count it advances and the version that count was read from.

type FirstFireClaim added in v0.22.0 

type FirstFireClaim func(ctx context.Context, ruleID, rowID string) bool

FirstFireClaim records that a created rule has fired for a row and reports whether THIS firing is the one that did: true for exactly one caller per (rule, row), on any replica. A false is a duplicate -- or a claim that could not be recorded, which is treated the same, because a created rule that cannot prove it is first must not send.

type FirstFireRelease added in v0.26.0 

type FirstFireRelease func(ctx context.Context, ruleID, rowID string)

FirstFireRelease gives back the claim a created rule took for a row, so the row's firing can be made again. It is for a firing that ended with nothing attempted -- the cluster's lock was busy -- and only that: a firing that began a send keeps its claim, which is what keeps the row from sending twice.

type Integration 

type Integration struct {
	// contains filtered or unexported fields
}

Integration is the provider the engine registers.

func (*Integration) Capabilities 

func (i *Integration) Capabilities() []memql.IntegrationCapability

func (*Integration) IntegrationName 

func (i *Integration) IntegrationName() string

type Result 

type Result struct {
	RuleID        string
	Status        string
	BundleID      string
	ConstructName string
	Lane          string
	Error         string
}

Result is what the builtin hands back.

type Rule 

type Rule struct {
	ID             string
	OwnerUserID    string
	Name           string
	Description    string
	TriggerConcept string
	EventKind      string
	Condition      string
	TemplateID     string
	RecipientMode  string
	RecipientRoles []string
	AudienceID     string
	RecipientField string
	AccountID      string
	SenderIdentity string
}

Rule is the form half of v1:campaigns:emailRule -- everything the generator reads. The generated/refs half (bundleId, constructName, lastError, the counters) is deliberately absent: those are the ENGINE's account of what it made of the form, and a generator that read them could produce a construct whose content depended on its own past output.

func (Rule) Validate 

func (r Rule) Validate() error

Validate checks the form for everything that can be decided without the live concept registry. What it deliberately does NOT check is whether the trigger concept exists and whether the recipient field resolves on it -- those need the registry, they belong to the activation path, and a generator that reached for a registry would stop being a pure function of the form.

type RuleState 

type RuleState struct {
	Status        string
	BundleID      string
	ConstructName string
	LastError     string
	FiredCount    int
	// VersionAt is the createdAt of the version read: the witness a record
	// conditional on this read is fenced on (memql.ContextWithRowVersionFence).
	// Zero when the read did not carry one.
	VersionAt time.Time
}

RuleState is the generated half of the row.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

func NewStore 

func NewStore(engine Engine) *Store

func (*Store) AddRecipient 

func (s *Store) AddRecipient(ctx context.Context, recipientID, audienceID, email, displayName, accountID, status string) error

AddRecipient enrols a row-derived address into the rule's audience, so it gains a membership, a subscription state and a working unsubscribe. `signup` is the honest source value: a product flow added it.

accountID is the rule's organization, as a BARE id -- the caller bare-ifies it once, because BareShortId is not idempotent (memql#2981). A recipient is an organization-scoped row and the send refuses one outside the campaign's organization, so an address enrolled without it is a recipient the very next step would refuse. A rule that names no organization (one that predates organizations) leaves it off the call, and addRecipient stamps "".

func (*Store) AudienceByID 

func (s *Store) AudienceByID(ctx context.Context, audienceID string) (bool, error)

AudienceByID reads the audience a marketing rule mails. Used at ARM time, not only at fire time: see Activator.verifyReadable.

func (*Store) AudienceConfirmation added in v0.26.0 

func (s *Store) AudienceConfirmation(ctx context.Context, id string) (string, error)

func (*Store) BundleStatus added in v0.26.0 

func (s *Store) BundleStatus(ctx context.Context, bundleID string) (status string, found bool, err error)

BundleStatus reads an authoring bundle's status under the caller's own actor. found is false when the bundle is not readable as the caller.

func (*Store) BundleVersion 

func (s *Store) BundleVersion(ctx context.Context, bundleID string) int

BundleVersion reads an authoring bundle's version. Zero when the bundle is unreadable -- which is treated as "there is nothing to supersede" rather than as an error, because the alternative is refusing to arm a rule on the strength of a row the caller cannot see.

func (*Store) CreateBundle 

func (s *Store) CreateBundle(ctx context.Context, bundleID, title, summary, supersedes string, version int) error

CreateBundle writes a rule's bundle as a draft. createAuthoringBundle is @serverOnly (memql#6022), so the caller stamps internal origin -- see Activator.writeBundle, where it is done inline at the one call.

func (*Store) CreateConstruct 

func (s *Store) CreateConstruct(ctx context.Context, constructID, bundleID, name, namespace, source string) error

func (*Store) EachSendableRecipient added in v0.26.0 

func (s *Store) EachSendableRecipient(ctx context.Context, audienceID string, visit func(map[string]any) bool) error

EachSendableRecipient follows the engine cursor and refuses a repeated cursor.

func (*Store) FiringStateByID added in v0.23.6 

func (s *Store) FiringStateByID(ctx context.Context, ruleID string) (FiringState, bool, error)

FiringStateByID reads the rule's count FRESH: from the database, never from this node's result cache. It is read inside the rule's firing gate, and the previous holder's write reaches this node's cache only by an asynchronous broadcast -- a cached answer is the count the previous holder already advanced (firing_gate.go, memql#5431).

The same named read as RuleStateByID, deliberately: freshness belongs to this one call site, and emailRuleById stays cached everywhere else it is read.

func (*Store) MailboxMembers added in v0.26.0 

func (s *Store) MailboxMembers(ctx context.Context, audienceID, address string) (members []map[string]any, complete bool, err error)

MailboxMembers returns every row of the rule's audience that holds the mailbox on the triggering row, in any subscription state, oldest first within each source -- and whether it can be sure that is all of them.

A row is found by the digest of the address (a read of the audience's rows that carry it) or, for the rows that carry no digest at all, by comparing its address (a page of them). Each row's OWN address is compared before the row is believed: a digest is a key, and a raw write can set one that is not its row's. This replaces a read that listed only the audience's subscribed members and stopped at the first hundred: it never saw anyone who had unsubscribed (so they were enrolled again as subscribed), and a mailbox beyond the hundred was enrolled a second time.

complete is false when EITHER page is full: there may be an opted-out row beyond it, so the caller must not treat the page as the whole mailbox. The campaigns package can fall back to the roster for rows without a digest; a rule has no roster read, so it refuses the firing and says why.

Every read here decides a write, so the caller's ctx must be a fresh read.

func (*Store) RecipientsForAudience 

func (s *Store) RecipientsForAudience(ctx context.Context, audienceID string) ([]map[string]any, error)

RecipientsForAudience reads the roster the marketing lane mails.

func (*Store) RecordDryRun 

func (s *Store) RecordDryRun(ctx context.Context, bundleID, ruleID, lane string) error

RecordDryRun writes Gate 2's verdict.

THE GATE IS PASSED BY CONSTRUCTION, and saying why matters. Gate 2 exists to put a behavioural trace and a side-effect manifest in front of a person before an LLM-authored bundle is armed. This bundle was not authored by an LLM: it is one automation with one step, produced by a deterministic template from a form the operator filled in and already saw, and the side effect it can have is exactly the one the form describes. Running a synthetic trace over it would manufacture an artifact nobody reads to satisfy a gate whose question has already been answered -- so the report says plainly what it is, and the operator's approval is the form they saved.

func (*Store) RecordFiring 

func (s *Store) RecordFiring(ctx context.Context, ruleID string, firedCount int, firedAt time.Time, lastError string) error

RecordFiring stamps that a rule ran. Its two fields are LIVENESS -- they move on their own -- which is why the surfaces that render them are told not to fingerprint them for arrival cues.

firedAt is both lastFiredAt and the new version's createdAt: the mutation stamps the version at it, which is what lets the fire path place its increment after the version it read (firedAtAfter).

func (*Store) RecordGeneration 

func (s *Store) RecordGeneration(ctx context.Context, ruleID, status, bundleID, constructName, lastError string) error

RecordGeneration stamps what the engine made of the form. @serverOnly, so the caller must have stamped internal origin -- see activate.go, where it is done inline at the one Execute that needs it.

func (*Store) RecordValidation 

func (s *Store) RecordValidation(ctx context.Context, bundleID string, ok bool, diagnostics []string, failureReason string) error

RecordValidation writes Gate 1's verdict. The report is an object literal rather than a quoted blob: `validationReport` is declared `object!`, and handing it a string would be a type the mutation does not accept.

func (*Store) RuleByID 

func (s *Store) RuleByID(ctx context.Context, ruleID string) (Rule, bool, error)

RuleByID reads one rule under the caller's own actor. That read IS the authorization on every path in this package: a rule the caller cannot read is a rule they cannot activate, retire or fire.

func (*Store) RuleOfBundle added in v0.26.0 

func (s *Store) RuleOfBundle(ctx context.Context, bundleID string) (ruleID, status string, ok bool, err error)

RuleOfBundle names the rule a generated bundle was built for, and the bundle's status, read under the caller's own actor. The bundle's dry-run report records the rule (RecordDryRun), and the bundle id is derived from the bundle's owner, that rule and the bundle's version (bundleIDFor) -- so the id is re-derived from what the row says, and a bundle whose row names a rule that does not derive it names none. A bundle armed before ids carried their owner is derived without it (legacyBundleIDFor), and is the rule's too.

func (*Store) RuleStateByID 

func (s *Store) RuleStateByID(ctx context.Context, ruleID string) (RuleState, bool, error)

RuleStateByID returns the fields the generator deliberately does not read -- the engine's own account of what it last made of the form.

func (*Store) SendToRecipient 

func (s *Store) SendToRecipient(ctx context.Context, templateID, recipientID, senderIdentityID, ruleID, requestID string) (map[string]any, error)

SendToRecipient is the MARKETING lane's one call: suppression checked at the point of send, the RFC 8058 pair attached, the resolved sending identity applied, the outcome ledgered. Reached as a builtin rather than as a Go call so this package does not depend on component/campaigns' internals.

A kind-prefixed `builtin <name>(...)` string IS executable through the engine, and the precedent is shipped: dsl/memql/tools.memql declares `@handler(type="query", query="builtin help(name: args.name)")`, and a query handler renders exactly such a call from its arguments' values and hands it to Execute (component/memql/tool_handler_v1.go).

func (*Store) SenderIdentityByID 

func (s *Store) SenderIdentityByID(ctx context.Context, identityID string) (map[string]any, bool, error)

SenderIdentityByID reads the identity a rule sends as.

func (*Store) StageOperationalEmail added in v0.26.0 

func (s *Store) StageOperationalEmail(ctx context.Context, requestID, templateID, to, displayName, dedupeKey, ruleID string) error

StageOperationalEmail renders the active rule's published template and stages it in the server outbox.

func (*Store) TemplateByID 

func (s *Store) TemplateByID(ctx context.Context, templateID string) (map[string]any, bool, error)

TemplateByID reads the copy both lanes send.

func (*Store) UsersInRoles 

func (s *Store) UsersInRoles(ctx context.Context) ([]map[string]any, error)

UsersInRoles reads the cluster people the operational lane mails.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.