source

package
v0.2.32-rc4 Latest Latest
Warning

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

Go to latest
Published: Oct 1, 2026 License: MPL-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package source is the chassis's outbound source-inlet seam: a cursor- bearing watcher that dials an EXTERNAL system a tenant delegated to us (today a remote IMAP mailbox; later Kafka/SQS/CDC/webhook-poll), reads what is new past a durable cursor, fires each item as a run of the tenant's `_source/0` stack, and — once that run succeeds — acknowledges the item at the source (marking it read, moving it aside).

This is the OPPOSITE direction from the `imap` personality, which is a server other people's mail clients connect INTO. A source connects OUT. It is also not an `EXEC` op: an op pulls inside one run, whereas a source is a durable subscription the chassis drives on a schedule, so its lifecycle (claim → poll → dispatch → ack → advance) lives in a poller personality, not on the request path.

The seam is deliberately kind-agnostic. A Kind opens connections to one class of external system; a Conn is one live attachment the poller owns for the duration of a claim. Adding a second kind (say Kafka) is a new Kind implementation that self-registers — no change to the poller, the store, the envelope shape, or the cursor plumbing, because the poller never interprets a Kind's cursor or item metadata.

Credentials never pass through this package as anything durable: the poller materializes the tenant's secret from the encrypted store, hands it to Open as a raw []byte, and zeroes it the instant Open returns. A Kind must not retain it — the authenticated Conn holds the socket, not the password. Nothing here is JSON-serializable in a way that could carry a credential into an envelope, trace, or log.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Kinds

func Kinds() []string

Kinds returns the registered kind names (for diagnostics / a "no such kind" error that lists what is available).

func RegisterKind

func RegisterKind(k Kind)

RegisterKind adds a source kind. Called from a kind package's init(); a duplicate name panics at boot (a programming error). The concrete kinds live in subpackages (imapsource) and are blank-imported by the binary, so a build that omits a kind simply cannot open a source of it.

func SourceID

func SourceID(tenant, stack, pack, declaredID string) string

SourceID is the deterministic primary key for a declared source: a hash of its natural key so a redeploy targets the same row (and its cursor) rather than minting a new one. srcseed and the store agree on it via this helper.

Types

type Action

type Action struct {
	Op   string // "none" | "seen" | "move"
	Dest string // destination mailbox/folder for "move"; empty otherwise
}

Action is what to do to an item at the source once its run succeeded. It is the parsed form of a source's `on_processed` config, overridable per run by the stack via `_txc.source.res.action`.

func ParseAction

func ParseAction(spec string) (Action, error)

ParseAction parses an action spec: "" / "none" (do nothing), "seen" (mark read), or "move:<dest>" (move aside to <dest>). A move with no destination is an error rather than a silent no-op, since it usually means a typo in config that would otherwise leave mail piling up unread.

func (Action) String

func (a Action) String() string

String renders an Action back to its spec form (round-trips ParseAction).

type Claimed

type Claimed struct {
	SourceID     string
	Tenant       string // slug
	Stack        string
	Pack         string
	DeclaredID   string
	Kind         string
	Config       json.RawMessage
	Cursor       Cursor // nil when never polled
	EverySeconds int
}

Claimed is a source this node won at ClaimDue: everything the poller needs to open the connection and route the fired runs.

type Conn

type Conn interface {
	// Poll reads items strictly after cur, up to limit, in ascending order.
	// Each returned Item carries the cursor to persist once that item's run
	// and ack both succeed. `drained` is the cursor to persist only if the
	// WHOLE batch is processed without a failure (or the batch is empty): it
	// carries side-band progress such as an IMAP UIDVALIDITY reset, so an
	// empty or renumbered source is not rescanned forever. drained may be nil.
	Poll(ctx context.Context, cur Cursor, limit int) (items []Item, drained Cursor, err error)

	// Ack applies act to one item at the source. Called by the poller only
	// after that item's run reached a successful terminal. A "none" action
	// is a no-op. An error stops the batch (the cursor does not advance past
	// this item), so Ack must be safe to retry on the next poll.
	Ack(ctx context.Context, it Item, act Action) error

	// Close releases the connection. Always called once per successful Open.
	Close() error
}

Conn is one live attachment to a source, owned by the poller for the duration of a single claim. It is NOT safe for concurrent use.

type Cursor

type Cursor json.RawMessage

Cursor is the durable "how far have I read" marker for one source. It is opaque to the poller and the store — only the Kind that produced it reads it (for IMAP it is `{"uidvalidity":N,"uid_hwm":M}`). A nil/empty Cursor means "from the beginning". Keeping it opaque is what lets a second kind with a completely different cursor shape reuse every other moving part.

type Declared

type Declared struct {
	Tenant       string // slug
	Stack        string
	Pack         string
	DeclaredID   string
	Kind         string
	Config       json.RawMessage // the pack line, verbatim
	Enabled      bool
	EverySeconds int
	Version      int64
}

Declared is the config-side of a source row — the columns srcseed owns. The runtime columns (cursor, claim, next_poll_at) are never in here, so an upsert of a Declared can never rewind a live cursor.

type Item

type Item struct {
	// Key is a stable dedup identity for this item within its source,
	// surfaced to the stack as `_txc.source.key` so a run can guard against
	// an at-least-once redelivery (kv/cas, or txco://imap/append object_key).
	// For IMAP it is "<uidvalidity>:<uid>".
	Key string

	// Raw is the item's bytes — RFC 5322 for an IMAP message. The poller
	// hands these to a kind-neutral parser (chassis/mail.ParseMessage for
	// mail) so a fetched message and an LMTP-delivered one produce identical
	// envelope JSON.
	Raw []byte

	// Meta is kind-specific facts the poller stamps under `_txc.source.*`
	// verbatim (it never interprets them) — e.g. `{"uid":42,"flags":[...]}`.
	Meta json.RawMessage

	// Cursor is the cursor to persist AFTER this item's run and ack both
	// succeed. Items arrive in order; the poller advances the stored cursor
	// item-by-item on success and stops at the first failure, so a mid-batch
	// failure never skips an unprocessed item.
	Cursor Cursor
}

Item is one unit of new input read from a source.

type Kind

type Kind interface {
	// Name is the kind identity used in a source's `kind` field ("imap").
	Name() string
	// Open dials and authenticates one connection. See OpenParams for the
	// credential lifetime contract.
	Open(ctx context.Context, p OpenParams) (Conn, error)
}

Kind opens connections to one class of external source.

func OpenKind

func OpenKind(name string) (Kind, bool)

OpenKind returns the registered kind, or (nil, false) if no kind of that name is built into this binary.

type OpenParams

type OpenParams struct {
	// Config is the kind-specific declared config blob (the pack line's
	// fields), e.g. `{"host":"imap.fastmail.com","port":993,...}`. It never
	// contains the secret VALUE — only its name, resolved by the poller.
	Config json.RawMessage

	// Cred is the materialized secret value (the mailbox password). It is
	// zeroed by the caller the moment Open returns; do not retain it.
	Cred []byte

	// Guard is the egress policy for the outbound dial. When non-nil a Kind
	// MUST route its connection through it (net.Dialer.Control =
	// egress.DialControl(guard)) so a source cannot be pointed at loopback,
	// link-local, or private address space. A nil Guard means "no policy"
	// (dev/test).
	Guard egress.Guard

	// Logger is the poller's logger, scoped for this source.
	Logger *zap.Logger
}

OpenParams carries everything a Kind needs to establish one Conn. The credential is live only for the duration of the Open call; a Kind must authenticate with it and not keep a reference.

type Status

type Status struct {
	SourceID   string `json:"source_id"`
	Stack      string `json:"stack"`
	Pack       string `json:"pack"`
	DeclaredID string `json:"declared_id"`
	Kind       string `json:"kind"`
	Enabled    bool   `json:"enabled"`
	Retired    bool   `json:"retired"`
	Status     string `json:"status"`
	ClaimedBy  string `json:"claimed_by,omitempty"`
	Cursor     string `json:"cursor,omitempty"`
	NextPollAt string `json:"next_poll_at,omitempty"`
	LastPollAt string `json:"last_poll_at,omitempty"`
	LastError  string `json:"last_error,omitempty"`
	Attempts   int    `json:"attempts"`
}

Status is one row's operator-visible state, for `txco source status`. It carries no secret — `config` is metadata (the secret is a name) and is omitted here; the value never existed in this table.

type Store

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

Store is the thin façade over the tenant_sources table (part of the shared RUNTIME DB, alongside tenant_secrets — not a separate backend, because a source's cursor and claim are runtime state that must live where the authoritative writes go). It carries the dialect for `?`→`$n` rebinding and the Postgres SKIP LOCKED claim, and a clock seam for tests, mirroring scheduled.Store and secrets.Store.

func NewStore

func NewStore(db *sql.DB, d registry.Dialect) *Store

NewStore builds a Store over the opened runtime DB and its dialect. A nil dialect defaults to SQLite (the in-tree default).

func (*Store) ClaimDue

func (s *Store) ClaimDue(ctx context.Context, node string, limit int) ([]Claimed, error)

ClaimDue claims up to limit due sources (enabled, live, unclaimed, and past next_poll_at) in ONE statement and returns the rows this call won — the same claim-is-the-coordination shape as scheduled.ClaimDue. On Postgres the due subselect takes FOR UPDATE SKIP LOCKED so concurrent pollers partition the due set; on SQLite the single statement is atomic under the one writer.

func (*Store) EnsureSchema

func (s *Store) EnsureSchema(ctx context.Context) error

EnsureSchema creates the tenant_sources table + indexes if absent. Prod gets the table from the runtime migration (0025); this exists so tests (and a belt on first boot) can stand the table up without the migration runner. The DDL is the portable subset the migration uses.

func (*Store) List

func (s *Store) List(ctx context.Context, tenant string) ([]Status, error)

List returns every source of one tenant (including retired), newest claim first, for the read-only status CLI.

func (*Store) ReclaimStale

func (s *Store) ReclaimStale(ctx context.Context, staleAfter time.Duration) (int64, error)

ReclaimStale returns sources stuck 'claimed' past staleAfter to 'idle' so another pass retries them — crash recovery for a node that died mid-poll. It leaves next_poll_at alone (it was already <= now), so the row is due at once. Returns the reset count.

func (*Store) Release

func (s *Store) Release(ctx context.Context, sourceID string, cursor Cursor, everySeconds int, pollErr error) error

Release ends a claim: it writes the advanced cursor (nil leaves it), records the outcome (last_error empty on success), stamps last_poll_at, schedules the next poll every_seconds out, and returns the row to 'idle'. Called once per claimed source after its poll pass — the mirror of ClaimDue.

func (*Store) RetireMissing

func (s *Store) RetireMissing(ctx context.Context, tenant, stack, pack string, keep []string) (int64, error)

RetireMissing soft-retires the rows of one (tenant, stack, pack) whose declared_id is NOT in keep — the sources a redeploy removed from a pack that is still present. Soft (retired_at set, row kept) so re-adding the source resumes from its cursor instead of re-reading the whole mailbox. An empty keep retires every row of that pack. Returns the retired count.

func (*Store) Upsert

func (s *Store) Upsert(ctx context.Context, d Declared) error

Upsert writes a declared source: it INSERTs the row (due immediately) or, if one already exists for (tenant, stack, pack, declared_id), UPDATEs ONLY the declared columns and clears retired_at (re-adding a source un-retires it and resumes from its stored cursor). The runtime columns — cursor, status, claim, next_poll_at — are deliberately absent from the UPDATE SET, so a redeploy never rewinds a live cursor or steals a claim.

Directories

Path Synopsis
Package imapsource is the IMAP source kind: it connects OUT to a remote IMAP mailbox a tenant delegated (Gmail, Fastmail, Dovecot, an Exchange- compatible server), reads messages past a durable UID cursor, and — once a message's run has succeeded — marks it read or moves it aside.
Package imapsource is the IMAP source kind: it connects OUT to a remote IMAP mailbox a tenant delegated (Gmail, Fastmail, Dovecot, an Exchange- compatible server), reads messages past a durable UID cursor, and — once a message's run has succeeded — marks it read or moves it aside.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL