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

retention

package
v1.9.0 Latest Latest
Warning

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

 Go to latest Published: Nov 2, 2026 License: MIT Imports: 35 Imported by: 0

Documentation

Overview

Package retention mirrors authsome auth activity into a CRM. Hooks write to an outbox and a background worker delivers, so a slow or unavailable CRM never shows up as login latency.

Each configured provider destination is either "generic" (a config-driven REST endpoint) or a named vendor implementation. Only "generic" ships today; a provider Type the plugin does not recognise fails at OnInit rather than dead-lettering every job later.

The generic provider's retry-classification policy (classifyHTTPError in provider_generic.go) is decided and recorded in "Retry classification" in docs/superpowers/specs/2026-09-03-crm-retention-delivery-design.md: one rule generates the whole table, that a failure affecting every job retries and a failure affecting only this job dies now.

The outbox is pruned rather than kept forever. The delivery worker sweeps terminal rows on its own ticker once an hour: done rows after 30 days, dead and suppressed rows after 180, all three configurable. Non-terminal rows are never eligible however old they are.

Deleting a done row releases its idempotency key, so the retention window is also the window in which a replayed hook is still deduplicated. Thirty days is chosen against that rather than against disk; see the Data model section of docs/superpowers/specs/2026-09-03-crm-retention-delivery-design.md for the reasoning.

Delivery is at-least-once. A repeated contact upsert is harmless because the contact ref turns it into an update; a repeated activity is not, so the job's idempotency key travels with the activity as its ExternalID and a provider that can key on it advertises CapActivityDedupe. HubSpot does, by searching for the id before it creates a second note. The generic provider sends the field and makes no promise, and the worker says so on every redelivery rather than leaving a duplicate nobody hears about.

Consent is optional. When the consent plugin is also registered and retention.require_consent is turned on, delivery is gated on an active grant for retention.consent_purpose; a user with no grant is never sent. Without the consent plugin, or with the setting left at its default, every user syncs.

Usage: 

eng, _ := authsome.NewEngine(
    authsome.WithStore(store),
    authsome.WithPlugin(retention.New(retention.Config{
        Providers: []retention.ProviderConfig{{
            Name:       "crm",
            Type:       "generic",
            ContactURL: "https://crm.example.com/api/contacts",
            Token:      os.Getenv("CRM_TOKEN"),
            FieldMap: map[string]string{
                "email":      "email_address",
                "first_name": "first_name",
                "last_name":  "last_name",
                "remote_id":  "id",
            },
        }},
        TickInterval: 15 * time.Second,
    })),
)

A GDPR export includes every CRM ref the user has, across all apps and providers, through ExportUserData: see export.go.

Index

Constants

View Source 
const (
	StatePending    = "pending"
	StateInFlight   = "in_flight"
	StateDone       = "done"
	StateDead       = "dead"
	StateSuppressed = "suppressed"
)

Outbox job states.

Suppressed is deliberately distinct from dead. "We chose not to send this" and "we tried and failed" are different answers when somebody audits you, and collapsing them loses the only record that a consent gate did its job.

View Source 
const (
	KindContactUpsert = "contact_upsert"
	KindActivityLog   = "activity_log"
)

Outbox job kinds.

Variables

View Source 
var (
	PostgresMigrations = migrate.NewGroup("authsome-retention", migrate.DependsOn("authsome"))
	SqliteMigrations   = migrate.NewGroup("authsome-retention", migrate.DependsOn("authsome"))
	MongoMigrations    = migrate.NewGroup("authsome-retention", migrate.DependsOn("authsome"))
)

Migration groups, one per driver. All depend on the core authsome group because both tables live in the same schema and the core group has to create it first. There is deliberately no foreign key from app_id to authsome_apps: see the design note for why the outbox does not carry one.

View Source 
var (
	// SettingEnabled turns delivery off without dropping queued work.
	//
	// The worker reads it per app immediately before delivering, so
	// flipping it off during a CRM incident stops the sends and leaves
	// the backlog claimable. See deliveryEnabled.
	SettingEnabled = settings.Define("retention.enabled", true,
		settings.WithDisplayName("CRM Retention Sync Enabled"),
		settings.WithDescription("Mirror signup and login activity into the configured CRM"),
		settings.WithCategory("Retention"),
		settings.WithScopes(settings.ScopeGlobal, settings.ScopeApp),
		settings.WithOrder(10),
	)

	// SettingRequireConsent gates delivery on an active consent grant.
	//
	// Defaults to false on purpose. Fail-closed reads as the responsible
	// default, and would be right if purposes were a fixed enum. They are
	// free text, so fail-closed means a fresh install watches the plugin do
	// nothing and files a bug.
	SettingRequireConsent = settings.Define("retention.require_consent", false,
		settings.WithDisplayName("Require Consent Before Sync"),
		settings.WithDescription("Only send to the CRM when the user has an active grant for the purpose below"),
		settings.WithCategory("Retention"),
		settings.WithScopes(settings.ScopeGlobal, settings.ScopeApp),
		settings.WithHelpText("Needs the consent plugin registered. With this on, a user with no grant is never sent."),
		settings.WithOrder(20),
	)

	// SettingConsentPurpose is the purpose string the gate checks.
	SettingConsentPurpose = settings.Define("retention.consent_purpose", "marketing",
		settings.WithDisplayName("Consent Purpose"),
		settings.WithDescription("The consent purpose that authorises CRM sync"),
		settings.WithCategory("Retention"),
		settings.WithScopes(settings.ScopeGlobal, settings.ScopeApp),
		settings.WithOrder(30),
	)
)
View Source 
var ErrNotFound = errors.New("retention: not found")

ErrNotFound is returned when a row does not exist. Every store maps its driver's own miss onto this, so callers never branch on a driver error.

Functions

func Retryable 

func Retryable(err error) (bool, time.Duration)

Retryable reports whether err asks to be retried, and after how long. An error that is not a ProviderError is treated as terminal on purpose: an unclassified failure retried forever is a queue that never drains, and the dead-letter row is the thing that gets it looked at.

Types

type Activity 

type Activity struct {
	Type       string            `json:"type"`
	OccurredAt time.Time         `json:"occurred_at"`
	Properties map[string]string `json:"properties,omitempty"`

	// ExternalID names this activity from our side, deterministically. It is
	// the outbox job's own idempotency key, so every delivery attempt at the
	// same job carries the same value and two different logins never share
	// one. A provider advertising CapActivityDedupe keys on it.
	//
	// Empty means the job carried no idempotency key, which the hooks never
	// produce. A provider must treat that as "cannot deduplicate" and create,
	// rather than matching every activity that also has no external id.
	ExternalID string `json:"external_id,omitempty"`

	// Redelivery says this job has already been through at least one
	// delivery attempt, so the activity may already exist in the CRM. It is
	// the signal a provider needs to decide whether the dedupe check is
	// worth its cost: on a first delivery there is nothing to collide with,
	// and HubSpot's search endpoint is rate limited to five requests per
	// second per account, so checking every time would halve the throughput
	// of the common case to protect the rare one.
	Redelivery bool `json:"redelivery,omitempty"`
}

Activity is one thing the user did, addressed to an existing contact.

type Capability 

type Capability uint8

Capability is a bitmask of what a CRM can actually accept. The worker checks it before enqueuing, so a provider is never handed a call it would have to fake success on. 

const (
	// CapContacts means the provider can create and update contacts.
	CapContacts Capability = 1 << iota
	// CapActivities means the provider can record an activity against a contact.
	CapActivities
	// CapActivityDedupe means LogActivity is idempotent on
	// Activity.ExternalID: given the same external id twice, the CRM ends up
	// with one activity entry, not two.
	//
	// It exists because delivery is at-least-once and contacts are protected
	// while activities are not. A contact ref makes a redelivered upsert an
	// update. Nothing plays that role for an activity, so a MarkDone that
	// fails after the provider call succeeded leaves the job in_flight, the
	// lease expires, and the CRM gets the same login logged twice.
	//
	// A provider that cannot promise this must not set the bit. The worker
	// then says so out loud when it redelivers, which is the honest answer:
	// the gap is real, and a silent nil would hide it the same way a
	// provider faking success on an unsupported call would.
	CapActivityDedupe
)

func (Capability) Has 

func (c Capability) Has(want Capability) bool

Has reports whether every bit in want is set.

type Config 

type Config struct {
	Providers    []ProviderConfig `json:"providers"`
	TickInterval time.Duration    `json:"tick_interval"` // default 30s
	Lease        time.Duration    `json:"lease"`         // default 2m
	BaseBackoff  time.Duration    `json:"base_backoff"`  // default 5s
	BatchSize    int              `json:"batch_size"`    // default 50
	MaxAttempts  int              `json:"max_attempts"`  // default 12

	// DoneRetention is how long a delivered row stays in the outbox.
	// Default 30 days. Set it to a negative duration to keep them forever.
	//
	// Deleting a done row releases its idempotency key, so this is also the
	// window in which a replayed hook is still deduplicated. Thirty days is
	// chosen against that, not against disk: a duplicate hook dispatch
	// happens within seconds of the original, and the outermost thing that
	// can re-present a job is its own retry budget, which tops out at
	// roughly 1.7 hours. See the Data model section of the spec.
	DoneRetention time.Duration `json:"done_retention"`

	// AuditRetention is how long a dead or suppressed row stays. Default
	// 180 days, six times DoneRetention, because these two are the audit
	// trail rather than the steady state. `suppressed` is the record that
	// the consent gate refused a send and `dead` is the record of what an
	// outage cost you; both are read after the fact, on a review or audit
	// cycle measured in quarters. They are also rare, so keeping them far
	// longer costs almost nothing. Negative keeps them forever.
	AuditRetention time.Duration `json:"audit_retention"`

	// PurgeInterval is how often the delivery worker sweeps expired rows.
	// Default 1 hour, which is 120 delivery ticks: the sweep shares the
	// worker's ticker rather than running a goroutine of its own, and it
	// must not run anywhere near delivery frequency. Non-positive disables
	// the sweep entirely.
	PurgeInterval time.Duration `json:"purge_interval"`
}

Config is the plugin's static configuration. Everything here has a working default except Providers, and an empty Providers list makes every hook a no-op rather than an error.

type Contact 

type Contact struct {
	UserID    id.UserID         `json:"user_id"`
	AppID     id.AppID          `json:"app_id"`
	Email     string            `json:"email"`
	FirstName string            `json:"first_name,omitempty"`
	LastName  string            `json:"last_name,omitempty"`
	Traits    map[string]string `json:"traits,omitempty"`
}

Contact is the normalized subset of a user that CRMs agree on.

type ContactRef 

type ContactRef struct {
	ID               id.RetentionRefID `json:"id"`
	AppID            id.AppID          `json:"app_id"`
	EnvID            id.EnvironmentID  `json:"env_id"`
	UserID           id.UserID         `json:"user_id"`
	Provider         string            `json:"provider"`
	RemoteObjectType string            `json:"remote_object_type"`
	RemoteID         string            `json:"remote_id"`
	SyncedAt         time.Time         `json:"synced_at"`
}

ContactRef records where a user landed in a given CRM. It is the only thing standing between a thousand logins and a thousand duplicate contacts.

func (*ContactRef) Ref 

func (c *ContactRef) Ref() RemoteRef

Ref converts the row into the shape providers speak.

type GenericProvider 

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

GenericProvider is the config-driven REST provider: it lets a CRM with no dedicated Go implementation work anyway, as long as it accepts a JSON POST and answers with a JSON body carrying the created/updated record's id.

func NewGenericProvider 

func NewGenericProvider(cfg ProviderConfig) (*GenericProvider, error)

NewGenericProvider builds a Provider from a ProviderConfig. ContactURL is required: Capabilities always advertises CapContacts, so a provider that could never actually reach a contacts endpoint must fail at construction rather than fail every delivery later.

func (*GenericProvider) Capabilities 

func (g *GenericProvider) Capabilities() Capability

Capabilities implements Provider. CapActivities is advertised only when ActivityURL is configured, so a CRM with no activity endpoint never gets handed an activity delivery it has no way to honour: the worker treats an unadvertised capability as a deliberate skip, not a failed send.

CapActivityDedupe is never advertised, and that is the honest answer rather than a gap waiting to be filled. This provider knows one endpoint and one field mapping. It cannot search the remote system, because it has no search URL, no query language and no result shape to read. It cannot assume the endpoint dedupes on external_id either: plenty of REST collections answer a repeated POST with a second row.

So the external id is sent (see activityFields) and the promise is not made. An operator whose CRM does upsert on that field gets the behaviour for free; everyone else gets a warning from the worker on every redelivery instead of a duplicate they never hear about. Claiming the bit to look complete is precisely the "works with any X" lie Capabilities was added to prevent.

func (*GenericProvider) LogActivity 

func (g *GenericProvider) LogActivity(ctx context.Context, ref RemoteRef, a *Activity) error

LogActivity implements Provider. It POSTs the activity, field-mapped, to ActivityURL, with the contact's remote id attached under the canonical field "contact_id" (itself subject to FieldMap, like every other field).

func (*GenericProvider) Name 

func (g *GenericProvider) Name() string

Name implements Provider.

func (*GenericProvider) UpsertContact 

func (g *GenericProvider) UpsertContact(ctx context.Context, c *Contact) (RemoteRef, error)

UpsertContact implements Provider. It POSTs the contact, field-mapped, to ContactURL, and reads the remote id out of the JSON response using the FieldMap["remote_id"] path (default "id").

type HubSpotProvider 

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

HubSpotProvider is the reference vendor Provider: a real CRM's contacts and notes (engagements) APIs mapped onto the two Provider methods.

func NewHubSpotProvider 

func NewHubSpotProvider(cfg ProviderConfig) (*HubSpotProvider, error)

NewHubSpotProvider builds a Provider talking to HubSpot's CRM API. Token is required: an empty token would otherwise silently send every request unauthenticated, and HubSpot's contacts and notes endpoints both require scoped auth, so that would only surface as a 401 on the first delivery instead of at startup.

func (*HubSpotProvider) Capabilities 

func (h *HubSpotProvider) Capabilities() Capability

Capabilities implements Provider. HubSpot's contacts and notes APIs cover contacts and activities unconditionally, unlike the generic provider where CapActivities depends on config.

CapActivityDedupe is claimed on the strength of LogActivity's search-before-create, with one honest limit: HubSpot's own docs say "it may take a few moments for newly created or updated CRM objects to appear in search results", so a redelivery that arrives inside that indexing lag can still create a second note. The window this exists to close is a failed MarkDone, which comes back only after the lease expires (two minutes by default), and that is well clear of a few moments. A redelivery driven by a failed attempt can come back in five seconds and is not.

func (*HubSpotProvider) LogActivity 

func (h *HubSpotProvider) LogActivity(ctx context.Context, ref RemoteRef, a *Activity) error

LogActivity implements Provider. It records the activity as a HubSpot note (the Notes/engagements API), associated to the contact via the HubSpot-defined note-to-contact association type.

On a redelivery it looks for the note first. Delivery is at-least-once, and where a repeated contact upsert is absorbed by the contact ref, nothing absorbs a repeated note: the CRM just gets the same login logged twice, in front of the CS team who asked for this data in the first place.

The search runs only when the worker says this job has been out before. A first delivery has nothing to collide with, and HubSpot rate limits the search endpoints to five requests per second per account, which is twenty times tighter than the ordinary object endpoints. Paying that on every login to protect the small fraction that get redelivered would cap the common case on the rarest one.

func (*HubSpotProvider) Name 

func (h *HubSpotProvider) Name() string

Name implements Provider.

func (*HubSpotProvider) UpsertContact 

func (h *HubSpotProvider) UpsertContact(ctx context.Context, c *Contact) (RemoteRef, error)

UpsertContact implements Provider. It searches for an existing contact by email first: HubSpot's create endpoint does not upsert on email for standard contacts, so creating unconditionally would produce a duplicate contact on every repeat signup/login for the same user.

type Job 

type Job struct {
	ID             id.RetentionJobID `json:"id"`
	AppID          id.AppID          `json:"app_id"`
	EnvID          id.EnvironmentID  `json:"env_id"`
	UserID         id.UserID         `json:"user_id"`
	Provider       string            `json:"provider"`
	Kind           string            `json:"kind"`
	Payload        map[string]string `json:"payload"`
	IdempotencyKey string            `json:"idempotency_key"`
	State          string            `json:"state"`
	Attempts       int               `json:"attempts"`
	NextAttemptAt  time.Time         `json:"next_attempt_at"`
	InFlightUntil  time.Time         `json:"in_flight_until"`
	LastError      string            `json:"last_error,omitempty"`
	CreatedAt      time.Time         `json:"created_at"`

	// Reclaimed is set by ClaimDue when this job was taken from an expired
	// lease rather than from pending. It is computed at claim time from the
	// row's state before the claim, and it is not a column: GetJob and
	// ListDead always report false.
	//
	// It exists because Attempts is blind to exactly the case that matters.
	// A delivery whose provider call succeeded and whose MarkDone then
	// failed leaves the row in_flight with Attempts unchanged, because
	// nothing got to record a failure. The lease expires, the row comes
	// back, and the only trace that it already went out once is that the
	// claim matched the expired-lease clause instead of the pending one.
	//
	// It cannot be recorded any earlier either. The failure it detects is a
	// store outage, and any marker we would write to warn ourselves goes to
	// the same store that just refused the write. The claim is the first
	// moment anything can see it.
	Reclaimed bool `json:"-"`
}

Job is one unit of pending delivery work.

func (*Job) Redelivered 

func (j *Job) Redelivered() bool

Redelivered reports whether this job may already have reached the provider once, so a delivery that is not naturally idempotent should check before repeating itself.

Two ways in. Attempts > 0 means a previous attempt reported an error, and an error is no proof the CRM did nothing: a request that succeeded and whose response we failed to read looks identical to one that never landed. Reclaimed means the previous attempt never reported anything at all.

type MemoryStore 

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

MemoryStore is a process-local Store, used in tests and when no database is configured. Nothing here survives a restart, which for this plugin means a pending backlog is lost rather than delayed.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore builds an empty in-memory store.

func (*MemoryStore) ClaimDue 

func (s *MemoryStore) ClaimDue(_ context.Context, limit int, lease time.Duration, now time.Time) ([]*Job, error)

func (*MemoryStore) DeleteRef 

func (s *MemoryStore) DeleteRef(_ context.Context, appID id.AppID, envID id.EnvironmentID,
	userID id.UserID, provider string) error

func (*MemoryStore) Enqueue 

func (s *MemoryStore) Enqueue(_ context.Context, j *Job) error

func (*MemoryStore) GetJob 

func (s *MemoryStore) GetJob(_ context.Context, jobID id.RetentionJobID) (*Job, error)

func (*MemoryStore) GetRef 

func (s *MemoryStore) GetRef(_ context.Context, appID id.AppID, envID id.EnvironmentID,
	userID id.UserID, provider string) (*ContactRef, error)

func (*MemoryStore) ListDead 

func (s *MemoryStore) ListDead(_ context.Context, appID id.AppID, limit int) ([]*Job, error)

func (*MemoryStore) ListRefsForUser 

func (s *MemoryStore) ListRefsForUser(_ context.Context, userID id.UserID) ([]*ContactRef, error)

ListRefsForUser returns every ref held for the user across all apps and providers. Deliberately unscoped by app: a data-subject export covers the person, not one app's view of them.

func (*MemoryStore) MarkDead 

func (s *MemoryStore) MarkDead(_ context.Context, jobID id.RetentionJobID, lastErr string) error

func (*MemoryStore) MarkDeferred 

func (s *MemoryStore) MarkDeferred(_ context.Context, jobID id.RetentionJobID, next time.Time, reason string) error

MarkDeferred returns the job to pending at next without spending an attempt. See the Store interface for why that distinction matters.

func (*MemoryStore) MarkDone 

func (s *MemoryStore) MarkDone(_ context.Context, jobID id.RetentionJobID, _ time.Time) error

func (*MemoryStore) MarkRetry 

func (s *MemoryStore) MarkRetry(_ context.Context, jobID id.RetentionJobID, next time.Time, lastErr string) error

func (*MemoryStore) MarkSuppressed 

func (s *MemoryStore) MarkSuppressed(_ context.Context, jobID id.RetentionJobID, reason string) error

func (*MemoryStore) PurgeTerminal 

func (s *MemoryStore) PurgeTerminal(_ context.Context, doneBefore, auditBefore time.Time) (int, error)

PurgeTerminal deletes terminal jobs older than their class cutoff. It also drops the row's idempotency key from the key index, which is the whole point: the key is only worth holding for as long as a replay of the hook that produced it is plausible, and the store's memory of it must go with the row rather than outlive it and block a legitimate re-enqueue forever.

func (*MemoryStore) PutRef 

func (s *MemoryStore) PutRef(_ context.Context, r *ContactRef) error

type MongoStore 

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

MongoStore implements Store on MongoDB. It shares neither models nor mappers with the SQL-backed stores: Mongo gets its own bson documents.

func NewMongoStore 

func NewMongoStore(db *grove.DB) *MongoStore

NewMongoStore builds a MongoDB-backed store.

func (*MongoStore) ClaimDue 

func (s *MongoStore) ClaimDue(ctx context.Context, limit int, lease time.Duration,
	now time.Time) ([]*Job, error)

ClaimDue loops FindOneAndUpdate with the due-or-expired filter, sorted by next_attempt_at then created_at, until it has limit documents or the filter stops matching. Each FindOneAndUpdate is atomic per document, which is what the contract needs: two callers racing on the same candidate cannot both claim it, because the second one's filter no longer matches once the first has flipped the state.

limit <= 0 means "no limit," matching every other backend: the loop then runs until the filter is exhausted instead of stopping at a count.

func (*MongoStore) DeleteRef 

func (s *MongoStore) DeleteRef(ctx context.Context, appID id.AppID, envID id.EnvironmentID,
	userID id.UserID, provider string) error

DeleteRef removes the contact ref. Deleting an absent ref is not an error.

func (*MongoStore) Enqueue 

func (s *MongoStore) Enqueue(ctx context.Context, j *Job) error

Enqueue inserts a pending job, treating a duplicate idempotency key as success. The partial unique index on idempotency_key (see MongoMigrations) is what makes this safe under concurrency.

func (*MongoStore) GetJob 

func (s *MongoStore) GetJob(ctx context.Context, jobID id.RetentionJobID) (*Job, error)

GetJob fetches one job. Returns ErrNotFound when absent.

func (*MongoStore) GetRef 

func (s *MongoStore) GetRef(ctx context.Context, appID id.AppID, envID id.EnvironmentID,
	userID id.UserID, provider string) (*ContactRef, error)

GetRef returns the contact ref. Returns ErrNotFound when absent.

func (*MongoStore) ListDead 

func (s *MongoStore) ListDead(ctx context.Context, appID id.AppID, limit int) ([]*Job, error)

ListDead returns dead-lettered jobs for an app, newest first.

func (*MongoStore) ListRefsForUser 

func (s *MongoStore) ListRefsForUser(ctx context.Context, userID id.UserID) ([]*ContactRef, error)

ListRefsForUser returns every ref held for the user across all apps and providers. Deliberately unscoped by app: a data-subject export covers the person, not one app's view of them.

func (*MongoStore) MarkDead 

func (s *MongoStore) MarkDead(ctx context.Context, jobID id.RetentionJobID, lastErr string) error

MarkDead parks a job permanently after too many attempts.

func (*MongoStore) MarkDeferred 

func (s *MongoStore) MarkDeferred(ctx context.Context, jobID id.RetentionJobID,
	nextAttemptAt time.Time, reason string) error

MarkDeferred returns a job to pending at nextAttemptAt without touching the attempt count. See the Store interface for why that is a separate method rather than a flag on MarkRetry.

func (*MongoStore) MarkDone 

func (s *MongoStore) MarkDone(ctx context.Context, jobID id.RetentionJobID, _ time.Time) error

MarkDone completes a job. now is intentionally unused: there is no completed_at field in the document.

func (*MongoStore) MarkRetry 

func (s *MongoStore) MarkRetry(ctx context.Context, jobID id.RetentionJobID,
	nextAttemptAt time.Time, lastErr string) error

MarkRetry returns a job to pending with an incremented attempt count.

func (*MongoStore) MarkSuppressed 

func (s *MongoStore) MarkSuppressed(ctx context.Context, jobID id.RetentionJobID, reason string) error

MarkSuppressed records that the job was deliberately not delivered.

func (*MongoStore) PurgeTerminal 

func (s *MongoStore) PurgeTerminal(ctx context.Context, doneBefore, auditBefore time.Time) (int, error)

PurgeTerminal deletes terminal documents older than their class cutoff.

A TTL index is the obvious Mongo-native alternative and it was rejected. Partial TTL indexes on an ordinary collection work fine -- expireAfterSeconds alongside partialFilterExpression has been supported since MongoDB 3.2 -- so "terminal states only" was never the blocker. What kills it: TTL indexes are single-field, and MongoDB refuses two indexes with the same key specification differing only in options, so `created_at` cannot carry both a 30-day window for done and a 180-day one for the audit states. Beyond that the reaper runs on its own roughly-60-second schedule, only on a primary, and would put the retention window in a migration constant instead of the plugin's config: Mongo would be the one backend whose purge behaviour the conformance suite cannot observe. An explicit DeleteMany keeps all four backends answering to the same contract.

func (*MongoStore) PutRef 

func (s *MongoStore) PutRef(ctx context.Context, r *ContactRef) error

PutRef inserts or updates the contact ref for the unique tuple. It looks up the existing document by the unique (app_id, env_id, user_id, provider) tuple first: an insert when absent, otherwise an update of the fields that can change.

Find-then-insert is not atomic, so two workers can both miss and both insert. That is likely rather than theoretical: a signup enqueues a contact_upsert and an activity_log for the same user in the same breath. The loser gets a duplicate key on the unique tuple index, so it falls back to the update instead of returning an error that would cost the job its retry budget for a document that is already where it should be.

func (*MongoStore) SetLogger 

func (s *MongoStore) SetLogger(l log.Logger)

SetLogger installs the plugin's logger. Optional: the store works without it, it just cannot tell anyone about a document it had to skip.

type Plugin 

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

Plugin mirrors signup/login activity into a CRM. See the package doc in provider.go for the outbox/worker shape.

func New 

func New(cfg ...Config) *Plugin

New builds the plugin. Config is optional.

logger is set to a no-op here, not left nil, because Task 7's hooks log an enqueue failure and can run before OnInit in tests. A nil logger there would turn a swallowed store error into a panic inside a login.

func (*Plugin) DeclareSettings 

func (p *Plugin) DeclareSettings(m *settings.Manager) error

DeclareSettings implements plugin.SettingsProvider.

func (*Plugin) ExportUserData 

func (p *Plugin) ExportUserData(ctx context.Context, userID id.UserID) (category string, data any, err error)

ExportUserData implements plugin.DataExportContributor. It returns every CRM ref held for the user, across all apps and providers, so a data subject can see exactly which external systems hold a copy of their record.

A nil store (the plugin registered but never initialised) returns an empty payload and no error: a user asking for their data should not get a 500 because a plugin was misconfigured.

func (*Plugin) MigrationGroups 

func (p *Plugin) MigrationGroups(driverName string) []*migrate.Group

MigrationGroups implements plugin.MigrationProvider.

func (*Plugin) Name 

func (p *Plugin) Name() string

Name returns the plugin name.

func (*Plugin) OnAfterSignIn 

func (p *Plugin) OnAfterSignIn(ctx context.Context, u *user.User, s *session.Session) error

OnAfterSignIn implements plugin.AfterSignIn.

func (*Plugin) OnAfterSignUp 

func (p *Plugin) OnAfterSignUp(ctx context.Context, u *user.User, s *session.Session) error

OnAfterSignUp implements plugin.AfterSignUp.

func (*Plugin) OnAfterUserUpdate 

func (p *Plugin) OnAfterUserUpdate(ctx context.Context, u *user.User) error

OnAfterUserUpdate implements plugin.AfterUserUpdate.

func (*Plugin) OnInit 

func (p *Plugin) OnInit(_ context.Context, engine plugin.Engine) error

OnInit captures engine references, picks a store for the driver in use, builds the configured providers and starts the delivery worker.

func (*Plugin) OnShutdown 

func (p *Plugin) OnShutdown(_ context.Context) error

OnShutdown stops the delivery worker. Safe to call on a plugin whose OnInit never ran, since worker is then nil.

func (*Plugin) RegisterProvider 

func (p *Plugin) RegisterProvider(pr Provider)

RegisterProvider adds or replaces a CRM provider by name. Tests use it to inject fakes; production wiring builds providers from Config in OnInit.

Safe to call at any time, including after OnInit and including while the delivery worker is running: enqueueFor and the worker both read providers through the same providerRegistry, so a provider registered after OnInit is enqueued for by the next hook and is deliverable by the worker's very next claim, with no restart and no window where it is one but not the other. Earlier this plugin handed the worker a private snapshot map taken at OnInit, so a late RegisterProvider was invisible to it: every job enqueued for that provider dead-lettered on first delivery attempt, permanently, because nothing here re-enqueues a dead row. See TestProviderRegisteredAfterOnInitReachesTheWorker.

func (*Plugin) SetStore 

func (p *Plugin) SetStore(s Store)

SetStore overrides the backing store. Tests use it; production wiring goes through OnInit.

type PostgresStore 

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

PostgresStore implements Store on the Grove ORM with PostgreSQL.

It shares jobModel, contactRefModel and every mapper with SqliteStore; only ClaimDue diverges, because it needs a single atomic statement that SQLite's conditional-update-then-check idiom cannot express.

func NewPostgresStore 

func NewPostgresStore(db *grove.DB) *PostgresStore

NewPostgresStore builds a PostgreSQL-backed store.

func (*PostgresStore) ClaimDue 

func (s *PostgresStore) ClaimDue(ctx context.Context, limit int, lease time.Duration,
	now time.Time) ([]*Job, error)

ClaimDue atomically moves up to limit due jobs to in_flight in one statement and returns them. See claimSQL for why the statement is shaped the way it is.

func (*PostgresStore) DeleteRef 

func (s *PostgresStore) DeleteRef(ctx context.Context, appID id.AppID, envID id.EnvironmentID,
	userID id.UserID, provider string) error

DeleteRef removes the contact ref. Deleting an absent ref is not an error.

func (*PostgresStore) Enqueue 

func (s *PostgresStore) Enqueue(ctx context.Context, j *Job) error

Enqueue inserts a pending job, treating a duplicate idempotency key as success. The unique index on idempotency_key is what makes this safe under concurrency; a read-then-write check would race two hooks firing at once.

func (*PostgresStore) GetJob 

func (s *PostgresStore) GetJob(ctx context.Context, jobID id.RetentionJobID) (*Job, error)

GetJob fetches one job. Returns ErrNotFound when absent.

func (*PostgresStore) GetRef 

func (s *PostgresStore) GetRef(ctx context.Context, appID id.AppID, envID id.EnvironmentID,
	userID id.UserID, provider string) (*ContactRef, error)

GetRef returns the contact ref. Returns ErrNotFound when absent.

func (*PostgresStore) ListDead 

func (s *PostgresStore) ListDead(ctx context.Context, appID id.AppID, limit int) ([]*Job, error)

ListDead returns dead-lettered jobs for an app, newest first.

func (*PostgresStore) ListRefsForUser 

func (s *PostgresStore) ListRefsForUser(ctx context.Context, userID id.UserID) ([]*ContactRef, error)

ListRefsForUser returns every ref held for the user across all apps and providers. Deliberately unscoped by app: a data-subject export covers the person, not one app's view of them.

func (*PostgresStore) MarkDead 

func (s *PostgresStore) MarkDead(ctx context.Context, jobID id.RetentionJobID, lastErr string) error

MarkDead parks a job permanently after too many attempts.

func (*PostgresStore) MarkDeferred 

func (s *PostgresStore) MarkDeferred(ctx context.Context, jobID id.RetentionJobID,
	nextAttemptAt time.Time, reason string) error

MarkDeferred returns a job to pending at nextAttemptAt without touching the attempt count. See the Store interface for why that is a separate method rather than a flag on MarkRetry.

func (*PostgresStore) MarkDone 

func (s *PostgresStore) MarkDone(ctx context.Context, jobID id.RetentionJobID, _ time.Time) error

MarkDone completes a job. now is intentionally unused: there is no completed_at column in the DDL.

func (*PostgresStore) MarkRetry 

func (s *PostgresStore) MarkRetry(ctx context.Context, jobID id.RetentionJobID,
	nextAttemptAt time.Time, lastErr string) error

MarkRetry returns a job to pending with an incremented attempt count.

func (*PostgresStore) MarkSuppressed 

func (s *PostgresStore) MarkSuppressed(ctx context.Context, jobID id.RetentionJobID, reason string) error

MarkSuppressed records that the job was deliberately not delivered.

func (*PostgresStore) PurgeTerminal 

func (s *PostgresStore) PurgeTerminal(ctx context.Context, doneBefore, auditBefore time.Time) (int, error)

PurgeTerminal deletes terminal rows older than their class cutoff, one statement per class. Same shape as SqliteStore.PurgeTerminal: three cheap statements running hourly, rather than one predicate whose AND/OR precedence a reader has to check before believing that a pending row is out of reach.

Deleting the row releases its idempotency key, because the unique index is on the table and not on a side ledger. That is deliberate and it is the property the retention window is chosen to make safe.

func (*PostgresStore) PutRef 

func (s *PostgresStore) PutRef(ctx context.Context, r *ContactRef) error

PutRef inserts or updates the contact ref for the unique tuple. It selects by the unique (app_id, env_id, user_id, provider) tuple first: an insert when absent, otherwise an update of the fields that can change.

Select-then-insert is not atomic, so two workers can both miss and both insert. That is likely rather than theoretical: a signup enqueues a contact_upsert and an activity_log for the same user in the same breath. The loser gets a unique violation on ux_retention_ref, so it falls back to the update instead of returning an error that would cost the job its retry budget for a row that is already exactly where it should be.

func (*PostgresStore) SetLogger 

func (s *PostgresStore) SetLogger(l log.Logger)

SetLogger installs the plugin's logger. Optional: the store works without it, it just cannot tell anyone about a row it had to skip.

type Provider 

type Provider interface {
	// Name is the provider's unique identifier, e.g. "hubspot".
	Name() string

	// Capabilities reports what this CRM can accept.
	Capabilities() Capability

	// UpsertContact creates or updates the contact and returns its ref.
	UpsertContact(ctx context.Context, c *Contact) (RemoteRef, error)

	// LogActivity records an activity against an existing contact.
	LogActivity(ctx context.Context, ref RemoteRef, a *Activity) error
}

Provider is a CRM the retention plugin can mirror auth activity into.

type ProviderConfig 

type ProviderConfig struct {
	Name        string            `json:"name"`
	Type        string            `json:"type"`
	Token       string            `json:"token,omitempty"`
	AuthType    string            `json:"auth_type,omitempty"`    // "bearer" (default) or "header"
	AuthHeader  string            `json:"auth_header,omitempty"`  // used when AuthType is "header"
	BaseURL     string            `json:"base_url,omitempty"`     // overridden in tests
	ContactURL  string            `json:"contact_url,omitempty"`  // generic only
	ActivityURL string            `json:"activity_url,omitempty"` // generic only; empty means no CapActivities
	FieldMap    map[string]string `json:"field_map,omitempty"`    // generic only
}

ProviderConfig configures one CRM destination. Type selects the implementation: "hubspot" for the vendor provider, "generic" for the config-driven REST one.

type ProviderError 

type ProviderError struct {
	Err        error
	Retryable  bool
	RetryAfter time.Duration // honoured when non-zero, e.g. from 429

	// DropRef means the remote record is gone, so the worker should delete
	// the local contact ref and let the retry recreate the contact rather
	// than keep updating something that no longer exists.
	DropRef bool

	// Status is the HTTP status the CRM answered with, or zero when the
	// request never landed. The worker logs its class (4xx, 5xx) in place
	// of the error text.
	Status int
}

ProviderError classifies a failure so the worker can decide retry vs dead-letter without knowing anything about the CRM's HTTP semantics.

func (*ProviderError) Error 

func (e *ProviderError) Error() string

func (*ProviderError) Unwrap 

func (e *ProviderError) Unwrap() error

type RemoteRef 

type RemoteRef struct {
	Provider   string `json:"provider"`
	ObjectType string `json:"object_type"`
	ID         string `json:"id"`
}

RemoteRef points at a record in the CRM. ObjectType is carried alongside the id because Salesforce-shaped CRMs need it before you can address the record.

func (RemoteRef) IsZero 

func (r RemoteRef) IsZero() bool

IsZero reports whether the ref names nothing.

type SqliteStore 

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

SqliteStore implements Store on the Grove ORM with SQLite.

func NewSqliteStore 

func NewSqliteStore(db *grove.DB) *SqliteStore

NewSqliteStore builds a SQLite-backed store.

func (*SqliteStore) ClaimDue 

func (s *SqliteStore) ClaimDue(ctx context.Context, limit int, lease time.Duration,
	now time.Time) ([]*Job, error)

ClaimDue claims due rows with a conditional update per row. The SELECT is only a candidate list: the UPDATE re-checks the due predicate and the row is kept only when it actually changed something, so two callers racing on the same candidate cannot both claim it.

A plain unconditional "UPDATE ... WHERE id = ?" would let both win. SQLite being single-writer does not help, because nothing stops two readers seeing the same row before either writes. This is the same conditional update + RowsAffected idiom used in store/sqlite/refresh_replay.go.

func (*SqliteStore) DeleteRef 

func (s *SqliteStore) DeleteRef(ctx context.Context, appID id.AppID, envID id.EnvironmentID,
	userID id.UserID, provider string) error

DeleteRef removes the contact ref. Deleting an absent ref is not an error.

func (*SqliteStore) Enqueue 

func (s *SqliteStore) Enqueue(ctx context.Context, j *Job) error

Enqueue inserts a pending job, treating a duplicate idempotency key as success. The unique index is what makes this safe under concurrency; a read-then-write check would race two hooks firing at once.

func (*SqliteStore) GetJob 

func (s *SqliteStore) GetJob(ctx context.Context, jobID id.RetentionJobID) (*Job, error)

GetJob fetches one job. Returns ErrNotFound when absent.

func (*SqliteStore) GetRef 

func (s *SqliteStore) GetRef(ctx context.Context, appID id.AppID, envID id.EnvironmentID,
	userID id.UserID, provider string) (*ContactRef, error)

GetRef returns the contact ref. Returns ErrNotFound when absent.

func (*SqliteStore) ListDead 

func (s *SqliteStore) ListDead(ctx context.Context, appID id.AppID, limit int) ([]*Job, error)

ListDead returns dead-lettered jobs for an app, newest first.

func (*SqliteStore) ListRefsForUser 

func (s *SqliteStore) ListRefsForUser(ctx context.Context, userID id.UserID) ([]*ContactRef, error)

ListRefsForUser returns every ref held for the user across all apps and providers. Deliberately unscoped by app: a data-subject export covers the person, not one app's view of them.

func (*SqliteStore) MarkDead 

func (s *SqliteStore) MarkDead(ctx context.Context, jobID id.RetentionJobID, lastErr string) error

MarkDead parks a job permanently after too many attempts.

func (*SqliteStore) MarkDeferred 

func (s *SqliteStore) MarkDeferred(ctx context.Context, jobID id.RetentionJobID,
	nextAttemptAt time.Time, reason string) error

MarkDeferred returns a job to pending at nextAttemptAt without touching the attempt count. See the Store interface for why that is a separate method rather than a flag on MarkRetry.

func (*SqliteStore) MarkDone 

func (s *SqliteStore) MarkDone(ctx context.Context, jobID id.RetentionJobID, _ time.Time) error

MarkDone completes a job. now is intentionally unused: there is no completed_at column in the DDL.

func (*SqliteStore) MarkRetry 

func (s *SqliteStore) MarkRetry(ctx context.Context, jobID id.RetentionJobID,
	nextAttemptAt time.Time, lastErr string) error

MarkRetry returns a job to pending with an incremented attempt count.

func (*SqliteStore) MarkSuppressed 

func (s *SqliteStore) MarkSuppressed(ctx context.Context, jobID id.RetentionJobID, reason string) error

MarkSuppressed records that the job was deliberately not delivered.

func (*SqliteStore) PurgeTerminal 

func (s *SqliteStore) PurgeTerminal(ctx context.Context, doneBefore, auditBefore time.Time) (int, error)

PurgeTerminal deletes terminal rows older than their class cutoff, one statement per class.

One statement per class rather than a single OR'd predicate so that the timestamp comparison is always the first bound parameter, which is what internal/sqliteguard checks, and so that no reader has to work out AND/OR precedence to see which rows a sweep can reach. It runs hourly by default; three round trips there cost nothing.

Deleting the row releases its idempotency key, because the unique index is on the table and not on a side ledger. That is deliberate and it is the property the retention window is chosen to make safe.

func (*SqliteStore) PutRef 

func (s *SqliteStore) PutRef(ctx context.Context, r *ContactRef) error

PutRef inserts or updates the contact ref for the unique tuple. It selects by the unique (app_id, env_id, user_id, provider) tuple first: an insert when absent, otherwise an update of the fields that can change.

Select-then-insert is not atomic, so two workers can both miss and both insert. That is likely rather than theoretical: a signup enqueues a contact_upsert and an activity_log for the same user in the same breath. The loser gets a unique violation on ux_retention_ref, so it falls back to the update instead of returning an error that would cost the job its retry budget for a row that is already exactly where it should be.

func (*SqliteStore) SetLogger 

func (s *SqliteStore) SetLogger(l log.Logger)

SetLogger installs the plugin's logger. Optional: the store works without it, it just cannot tell anyone about a row it had to skip.

type Store 

type Store interface {
	// Enqueue inserts a pending job. Inserting a job whose IdempotencyKey
	// already exists is a no-op and returns nil, so a double-fired hook
	// cannot produce two deliveries.
	Enqueue(ctx context.Context, j *Job) error

	// ClaimDue atomically moves up to limit due pending jobs to in_flight and
	// returns them, setting InFlightUntil to now.Add(lease).
	//
	// A job is due when state is pending and NextAttemptAt is at or before
	// now, OR when state is in_flight and InFlightUntil has passed. That
	// second clause is what recovers work from a process that died mid
	// delivery; without it those rows are invisible to every later claim and
	// the user behind them silently stops syncing.
	//
	// A job claimed through that second clause comes back with Reclaimed
	// set, which is the only signal anywhere that it may already have been
	// delivered once. See Job.Reclaimed.
	ClaimDue(ctx context.Context, limit int, lease time.Duration, now time.Time) ([]*Job, error)

	// MarkDone completes a job.
	MarkDone(ctx context.Context, jobID id.RetentionJobID, now time.Time) error

	// MarkRetry returns a job to pending with an incremented attempt count.
	MarkRetry(ctx context.Context, jobID id.RetentionJobID, nextAttemptAt time.Time, lastErr string) error

	// MarkDeferred returns a job to pending at nextAttemptAt WITHOUT
	// incrementing the attempt count, and records reason in the same field
	// MarkRetry uses for its last error.
	//
	// It exists for the case where nothing failed and nothing was decided:
	// delivery is switched off for the job's app, so the job goes back on
	// the queue untouched. MarkRetry would spend an attempt, and a long
	// enough disable would then burn the whole budget for jobs that never
	// failed, so the first real error after re-enabling would dead-letter
	// them immediately.
	MarkDeferred(ctx context.Context, jobID id.RetentionJobID, nextAttemptAt time.Time, reason string) error

	// MarkDead parks a job permanently after too many attempts.
	MarkDead(ctx context.Context, jobID id.RetentionJobID, lastErr string) error

	// MarkSuppressed records that the job was deliberately not delivered.
	MarkSuppressed(ctx context.Context, jobID id.RetentionJobID, reason string) error

	// PurgeTerminal deletes terminal rows that are older than their cutoff
	// and returns how many went. `done` rows are compared against
	// doneBefore; `dead` and `suppressed` against auditBefore, which is the
	// longer of the two because those two are the audit trail.
	//
	// Only terminal rows are eligible. A `pending` or `in_flight` row is
	// never deleted however old it is: age there means a stuck job, and
	// deleting it destroys work nobody knows is missing.
	//
	// The comparison is on CreatedAt, because there is no completed_at
	// column and adding one to buy a few hours of precision against a
	// window measured in months is not worth the migration. A job that
	// spent its whole retry budget dying therefore ages out about 1.7
	// hours "early" relative to the moment it actually died.
	//
	// A zero cutoff means "delete nothing in that class", which is how a
	// non-positive retention setting switches the sweep off. Every backend
	// gets that for free: nothing was created before year one.
	//
	// Deleting a `done` row releases its idempotency key. See the
	// replay-window note in the Data model section of
	// docs/superpowers/specs/2026-09-03-crm-retention-delivery-design.md
	// for why that is safe at these distances and not at shorter ones.
	PurgeTerminal(ctx context.Context, doneBefore, auditBefore time.Time) (int, error)

	// GetJob fetches one job. Returns ErrNotFound when absent.
	GetJob(ctx context.Context, jobID id.RetentionJobID) (*Job, error)

	// ListDead returns dead-lettered jobs for an app, newest first.
	ListDead(ctx context.Context, appID id.AppID, limit int) ([]*Job, error)

	// GetRef returns the contact ref. Returns ErrNotFound when absent.
	GetRef(ctx context.Context, appID id.AppID, envID id.EnvironmentID,
		userID id.UserID, provider string) (*ContactRef, error)

	// PutRef inserts or updates the contact ref for the unique tuple.
	PutRef(ctx context.Context, r *ContactRef) error

	// DeleteRef removes the contact ref, so the next attempt recreates the
	// contact. Deleting an absent ref is not an error.
	DeleteRef(ctx context.Context, appID id.AppID, envID id.EnvironmentID,
		userID id.UserID, provider string) error

	// ListRefsForUser returns every CRM ref held for one user across all
	// providers, for the data export. Returns an empty slice, not an error,
	// when the user has none.
	ListRefsForUser(ctx context.Context, userID id.UserID) ([]*ContactRef, error)
}

Store persists outbox jobs and contact refs.

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.