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

store

package
v0.4.1 Latest Latest
Warning

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

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

Documentation

Overview

Package store is the platform's durable operational store on PostgreSQL (design §13.1). Every query is scoped by organisation, and by memory store where records are involved. Uniqueness, deduplication and optimistic concurrency are enforced by database constraints; fenced writes check the caller's lease generation inside the same transaction (ADR-0005).

Index

Constants

View Source 
const (
	DeliveryLease        = 15 * time.Minute
	MaxAttempts          = 5
	MaxPendingDeliveries = 1000
	MaxEventBytes        = 8 << 10
)

Inbox limits (contracts.md, ADR-0006).

View Source 
const (
	KindNote = "note"
	KindWork = "work"
)

Record kinds.

View Source 
const (
	OriginHuman    = "human"
	OriginSeat     = "seat"
	OriginSystem   = "system"
	OriginSchedule = "schedule"
	OriginProbe    = "probe"
)

Message origins. They are assigned by trusted code only (§9.2).

View Source 
const (
	ConvHuman    = "human"
	ConvInternal = "internal"
	ConvSystem   = "system"
)

Conversation kinds.

View Source 
const (
	OpPending   = "pending"
	OpSucceeded = "succeeded"
	OpFailed    = "failed"
	OpUnknown   = "unknown"
)

Operation statuses (§10.3).

View Source 
const (
	OutboxPending = "pending"
	OutboxSending = "sending"
	OutboxSent    = "sent"
	OutboxUnknown = "unknown"
	OutboxDead    = "dead"
)

Outbox states. A claimed row is "sending" until resolved; a claim that is never resolved (crash mid-send) becomes unknown, never a blind resend.

View Source 
const (
	PubPending   = "pending"
	PubPublished = "published"
	// PubBlocked: the connection is unavailable; retried with backoff. It
	// never affects the work item itself.
	PubBlocked = "blocked"
	PubFailed  = "failed"
)

Work publication states.

View Source 
const DefaultRetirementGrace = 10 * time.Minute

DefaultRetirementGrace bounds a retiring seat's wind-down.

View Source 
const LeaseTTL = 30 * time.Second

LeaseTTL is the execution lease lifetime without renewal (contracts.md).

View Source 
const MaxConsoleEvents = 2000

MaxConsoleEvents bounds the events returned for one run.

View Source 
const MaxPassivePerTurn = 20

MaxPassivePerTurn bounds the passive messages handed over with one turn.

View Source 
const MaxSchedulesPerSeat = 50

MaxSchedulesPerSeat bounds a seat's active wake schedules.

Variables

View Source 
var (
	ErrNotFound = errors.New("not found")
	// ErrFenced means the caller's lease generation is not the current one.
	ErrFenced   = errors.New("fenced")
	ErrConflict = errors.New("conflict")
	ErrInvalid  = errors.New("invalid")
	// ErrInboxFull means the recipient already has MaxPendingDeliveries (ADR-0006).
	ErrInboxFull = errors.New("recipient inbox full")
)
View Source 
var ErrRecipientRetiring = errors.New("recipient seat is retiring")

ErrRecipientRetiring means the recipient seat is retiring or retired and accepts no new messages.

View Source 
var ErrStructured = errors.New("structured record")

ErrStructured means a plain memory write targeted a structured record.

Functions

func ExternalRef 

func ExternalRef(channel, thread string) string

ExternalRef encodes an external channel and thread as a conversation ref.

func SplitExternalRef 

func SplitExternalRef(ref string) (channel, thread string)

SplitExternalRef reverses ExternalRef.

Types

type ConsoleOperationFilter 

type ConsoleOperationFilter struct {
	SeatKey string
	Status  string
	Limit   int
}

ConsoleOperationFilter narrows ConsoleOperations.

type ConversationSummary 

type ConversationSummary struct {
	ID        string    `json:"conversation_id"`
	Kind      string    `json:"kind"`
	Binding   string    `json:"binding,omitempty"`
	CreatedAt time.Time `json:"created_at"`
	LastAt    time.Time `json:"last_message_at"`
}

ConversationSummary describes a conversation the seat takes part in.

type DueRetirement added in v0.3.0 

type DueRetirement struct {
	OrganizationID string
	SeatID         string
	Key            string
	// Overdue means the grace period ended; otherwise the seat finished its
	// retirement turn and has nothing left in flight.
	Overdue bool
}

DueRetirement is a retiring seat ready to be retired.

type Entry added in v0.2.0 

type Entry struct {
	RecordID  string    `json:"record_id"`
	Store     string    `json:"store"`
	Path      string    `json:"path"`
	Kind      string    `json:"kind"`
	Revision  int       `json:"revision"`
	Bytes     int       `json:"bytes"`
	Author    string    `json:"author_seat"`
	CreatedAt time.Time `json:"created_at"`
	UpdatedAt time.Time `json:"updated_at"`
}

Entry is a record's listing line (no body).

type Fence 

type Fence struct {
	SeatID     string
	Generation int64
}

Fence identifies a seat's write and the lease generation it claims.

type FileWrite added in v0.2.0 

type FileWrite struct {
	OrganizationID string
	StoreID        string
	AuthorID       string
	Path           string
	Text           string
	Tags           []string
	SourceRefs     []string
	// ExpectedRevision, when set, makes the write a compare-and-swap: 0
	// means the path must not exist yet.
	ExpectedRevision *int
}

FileWrite creates or replaces the note at a path.

type Gauges 

type Gauges struct {
	// SeatsByState counts active seats per organisation key and state.
	SeatsByState map[[2]string]int
	// OldestPending is the oldest undelivered message age per organisation key.
	OldestPending map[string]float64
}

Gauges is a snapshot for the §14 operational gauges.

type HumanMessage 

type HumanMessage struct {
	OrganizationID string
	Connection     string
	Binding        string
	ExternalUserID string
	// ExternalRef identifies the external conversation (channel and thread).
	ExternalRef      string
	EventID          string
	RepresentativeID string
	Body             string
}

HumanMessage is a verified inbound event for a representative.

type HumanReply 

type HumanReply struct {
	OrganizationID string
	SeatID         string
	Binding        string
	Connection     string
	// ConversationID is the human conversation replied to; when empty the
	// representative's latest conversation for the binding is used.
	ConversationID string
	ParentID       string
	Body           string
}

HumanReply is an outbound message from a representative to its bound human.

type Ingested 

type Ingested struct {
	MessageID string
	Duplicate bool
	// OverCap reports that the representative's inbox exceeded its cap. The
	// message is still accepted: accepted human messages are never dropped.
	OverCap bool
}

Ingested is the outcome of IngestHuman.

type ListOrder added in v0.2.0 

type ListOrder string

ListOrder selects the listing order column. 

const (
	OrderName      ListOrder = "name"
	OrderCreatedAt ListOrder = "created_at"
	OrderUpdatedAt ListOrder = "updated_at"
)

Listing orders.

type Message 

type Message struct {
	ID               string    `json:"message_id"`
	ConversationID   string    `json:"conversation_id"`
	ConversationKind string    `json:"-"`
	Binding          string    `json:"binding,omitempty"`
	Participants     []string  `json:"-"`
	Origin           string    `json:"origin"`
	SenderSeatID     string    `json:"-"`
	SenderSeat       string    `json:"sender_seat,omitempty"`
	RecipientSeatID  string    `json:"-"`
	RecipientSeat    string    `json:"recipient_seat,omitempty"`
	RecipientBinding string    `json:"recipient_binding,omitempty"`
	ParentID         string    `json:"parent_id,omitempty"`
	CorrelationID    string    `json:"correlation_id,omitempty"`
	Body             string    `json:"body"`
	CreatedAt        time.Time `json:"created_at"`
}

Message is a stored message with its conversation context.

type NewOperation 

type NewOperation struct {
	OrganizationID string
	SeatID         string
	ExecutionID    string
	Connection     string
	Operation      string
	Target         string
	RequestHash    string
	IdempotencyKey string
}

NewOperation is a connector side effect about to be attempted.

type NextFunc 

type NextFunc func(schedule string, fired, now time.Time) (next time.Time, ok bool)

NextFunc computes a schedule's next trigger after firing at fired; ok=false means the schedule is finished.

type Organization 

type Organization struct {
	ID        string
	Key       string
	Namespace string
	Manifest  compile.Manifest
}

Organization is an active organisation and its effective manifest.

type OutboxItem 

type OutboxItem struct {
	ID             int64
	OrganizationID string
	MessageID      string
	OperationID    string
	Connection     string
	Binding        string
	ExternalRef    string
	Body           string
	Attempts       int
}

OutboxItem is a claimed outbound message.

type Publication added in v0.2.0 

type Publication struct {
	RecordID          string    `json:"record_id"`
	Path              string    `json:"path"`
	Store             string    `json:"store"`
	Connection        string    `json:"connection"`
	ExternalID        string    `json:"external_id,omitempty"`
	ExternalURL       string    `json:"external_url,omitempty"`
	State             string    `json:"state"`
	PublishedRevision int       `json:"published_revision"`
	Revision          int       `json:"revision"`
	Attempts          int       `json:"attempts"`
	LastError         string    `json:"last_error,omitempty"`
	UpdatedAt         time.Time `json:"updated_at"`
}

Publication is the status of one work item's publication.

type PublicationItem added in v0.2.0 

type PublicationItem struct {
	OrganizationID    string
	RecordID          string
	Connection        string
	ExternalKind      string
	ExternalID        string
	ExternalURL       string
	PublishedRevision int
	Attempts          int
	Revision          int
	Data              json.RawMessage
	Body              string
}

PublicationItem is a claimed publication with the work item to publish.

type Record 

type Record struct {
	ID         string          `json:"record_id"`
	Store      string          `json:"store"`
	StoreID    string          `json:"-"`
	Revision   int             `json:"revision"`
	Path       string          `json:"path"`
	Kind       string          `json:"kind"`
	Data       json.RawMessage `json:"-"`
	Text       string          `json:"text"`
	Tags       []string        `json:"tags,omitempty"`
	SourceRefs []string        `json:"source_refs,omitempty"`
	Author     string          `json:"author_seat"`
	Archived   bool            `json:"archived,omitempty"`
	CreatedAt  time.Time       `json:"created_at"`
	UpdatedAt  time.Time       `json:"updated_at"`
}

Record is the current revision of a memory record.

type Retirement added in v0.3.0 

type Retirement struct {
	OrganizationID string
	SeatID         string
	Key            string
	Overdue        bool
	// Interrupted counts executions still running at the deadline.
	Interrupted int
	// Returned are the messages returned to their senders at retirement.
	Returned int
	// ReleasedWork are the work item ids released back to ready.
	ReleasedWork []string
	// Notified are the representative seats sent the summary.
	Notified []string
}

Retirement reports what a finished retirement handed back.

type RevisionConflict 

type RevisionConflict struct {
	Current       int
	CurrentAuthor string
}

RevisionConflict reports a stale expected revision (A11).

func IsConflict 

func IsConflict(err error) (*RevisionConflict, bool)

IsConflict extracts a revision conflict.

func (*RevisionConflict) Error 

func (e *RevisionConflict) Error() string

func (*RevisionConflict) Is 

func (e *RevisionConflict) Is(target error) bool

type RevisionInfo 

type RevisionInfo struct {
	Revision   int       `json:"revision"`
	Path       string    `json:"path"`
	Author     string    `json:"author_seat"`
	Generation *int64    `json:"author_generation,omitempty"`
	SourceRefs []string  `json:"source_refs,omitempty"`
	Archived   bool      `json:"archived,omitempty"`
	BodyBytes  int       `json:"bytes"`
	CreatedAt  time.Time `json:"created_at"`
}

RevisionInfo is one entry of a record's history (body omitted).

type Schedule 

type Schedule struct {
	ID            string     `json:"id"`
	Schedule      string     `json:"schedule"`
	Note          string     `json:"note,omitempty"`
	NextTriggerAt time.Time  `json:"next_trigger_at"`
	LastFiredAt   *time.Time `json:"last_fired_at,omitempty"`
	CreatedAt     time.Time  `json:"created_at"`
}

Schedule is a durable wake trigger (§9.3).

type SearchHit 

type SearchHit struct {
	RecordID  string    `json:"record_id"`
	Store     string    `json:"store"`
	Path      string    `json:"path"`
	Kind      string    `json:"kind"`
	Snippet   string    `json:"snippet"`
	Tags      []string  `json:"tags,omitempty"`
	Revision  int       `json:"revision"`
	Author    string    `json:"author_seat"`
	UpdatedAt time.Time `json:"updated_at"`
}

SearchHit is a bounded, attributed full-text search result.

type Seat 

type Seat struct {
	ID             string
	OrganizationID string
	Key            string
	ConfigRevision string
	PolicyRevision int64
	Manifest       compile.SeatManifest
	// RetireBy is set while the seat is retiring: it takes no new messages
	// and winds down until then.
	RetireBy *time.Time
}

Seat is an active seat with its current policy revision.

type SeatMessage 

type SeatMessage struct {
	OrganizationID string
	SenderID       string
	RecipientID    string
	// ConversationID continues an internal conversation the sender takes part in.
	ConversationID string
	ParentID       string
	CorrelationID  string
	Route          string
	Body           string
	// Passive queues the message without starting a turn: it is handed over
	// with the recipient's next turn, whatever starts it.
	Passive bool
}

SeatMessage is an internal message between seats. Route authorisation is the caller's responsibility; the store enforces conversation membership.

type Sent 

type Sent struct {
	MessageID      string `json:"message_id"`
	ConversationID string `json:"conversation_id"`
	CorrelationID  string `json:"correlation_id"`
}

Sent identifies an accepted message.

type Store 

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

Store wraps a pgx connection pool.

func Open 

func Open(ctx context.Context, url string) (*Store, error)

Open connects to the database at url and applies pending migrations.

func (*Store) Ack 

func (s *Store) Ack(ctx context.Context, f Fence, deliveryID int64, req runtimeapi.InboxAckRequest) error

Ack completes a leased delivery and its execution. A failed outcome counts as an attempt and is retried with backoff until dead-lettered.

func (*Store) AcquireLease 

func (s *Store) AcquireLease(ctx context.Context, seatID, podUID string) (*runtimeapi.LeaseResponse, error)

AcquireLease grants the seat's execution lease to podUID when it is expired, unheld or already held by podUID, incrementing the generation. Deliveries leased by an older generation go back to the queue as a failed attempt.

func (*Store) ActiveOrganizations 

func (s *Store) ActiveOrganizations(ctx context.Context) ([]string, error)

ActiveOrganizations lists the ids of all non-deleted organisations.

func (*Store) ActiveStoreIDs 

func (s *Store) ActiveStoreIDs(ctx context.Context, orgID string, keys []string) (map[string]string, error)

ActiveStoreIDs maps the given store keys to the ids of their active stores. Unknown or retired keys are omitted.

func (*Store) AppendEvents 

func (s *Store) AppendEvents(ctx context.Context, f Fence, executionID string, events []runtimeapi.ExecutionEvent) error

AppendEvents records execution events for one of the seat's executions. Oversized payloads are replaced by a marker (ADR-0006).

func (*Store) AppendFile added in v0.2.0 

func (s *Store) AppendFile(ctx context.Context, f Fence, orgID, storeID, authorID, path, text string, maxBytes int) (*Record, error)

AppendFile appends text to the note at path, creating it when missing. It fails with ErrInvalid when the result would exceed maxBytes.

func (*Store) ArchiveRecord added in v0.2.0 

func (s *Store) ArchiveRecord(ctx context.Context, f Fence, orgID, recordID, authorID string, expected int, storeIDs []string) (*Record, error)

ArchiveRecord hides a note from normal retrieval if it is still at the expected revision. Structured records are not archived this way.

func (*Store) BeginOperation 

func (s *Store) BeginOperation(ctx context.Context, f Fence, in NewOperation) (op *runtimeapi.Operation, created bool, err error)

BeginOperation records a pending operation before its side effect, in a fenced transaction. If the idempotency key was already used, the existing operation is returned with created=false and must not be re-executed.

func (*Store) BeginSystemOperation added in v0.2.0 

func (s *Store) BeginSystemOperation(ctx context.Context, in NewOperation) (*runtimeapi.Operation, bool, error)

BeginSystemOperation records a pending platform-initiated operation (no seat, no lease fence). A reused idempotency key returns the recorded operation with created=false; it must not be re-executed blindly.

func (*Store) CancelSchedule 

func (s *Store) CancelSchedule(ctx context.Context, f Fence, id string) error

CancelSchedule deactivates one of the seat's schedules.

func (*Store) Checkpoints 

func (s *Store) Checkpoints(ctx context.Context, seatID string) ([]runtimeapi.Checkpoint, error)

Checkpoints returns the seat's checkpoints, most recent first.

func (*Store) ClaimOutbox 

func (s *Store) ClaimOutbox(ctx context.Context, limit int) ([]OutboxItem, error)

ClaimOutbox claims up to limit due outbound messages and records a pending connector operation for each before anything is sent.

func (*Store) ClaimPublications added in v0.2.0 

func (s *Store) ClaimPublications(ctx context.Context, orgID, connection string, limit int, lease time.Duration) ([]PublicationItem, error)

ClaimPublications reserves, for lease, up to limit publications that are due: never published, or behind the work item's current revision. A publisher that dies holding a claim is retried once the lease ends.

func (*Store) Close 

func (s *Store) Close()

Close releases the pool.

func (*Store) ConsoleActivity 

func (s *Store) ConsoleActivity(ctx context.Context, orgID string, after time.Time, limit int) ([]runtimeapi.ConsoleActivityItem, error)

ConsoleActivity returns the organisation's activity after the given time, oldest first: messages, runs starting and finishing, tool calls, harness errors and connector operation status changes. With a zero after it returns the most recent items instead.

func (*Store) ConsoleArtifacts 

func (s *Store) ConsoleArtifacts(ctx context.Context, orgID string, limit int) ([]runtimeapi.ConsoleArtifact, error)

ConsoleArtifacts lists the organisation's artifacts, newest first.

func (*Store) ConsoleConversation 

func (s *Store) ConsoleConversation(ctx context.Context, id string) (*runtimeapi.ConsoleConversationDetail, error)

ConsoleConversation returns a conversation with its messages and their deliveries.

func (*Store) ConsoleConversations 

func (s *Store) ConsoleConversations(ctx context.Context, orgID string, includeSystem bool, limit int) ([]runtimeapi.ConsoleConversation, error)

ConsoleConversations lists the organisation's conversations, most recently active first. System conversations (probes, schedules) are included only when asked for.

func (*Store) ConsoleExecution 

func (s *Store) ConsoleExecution(ctx context.Context, id string) (*runtimeapi.ConsoleExecutionDetail, error)

ConsoleExecution returns a run with its trigger message, events, connector operations and the messages its seat sent under that run's lease.

func (*Store) ConsoleExecutions 

func (s *Store) ConsoleExecutions(ctx context.Context, seatID string, before time.Time, limit int) ([]runtimeapi.ConsoleExecution, error)

ConsoleExecutions pages a seat's runs, newest first, started before before (zero for the latest). Runs from every lease generation are included, so the history outlives any one Pod.

func (*Store) ConsoleOperations 

func (s *Store) ConsoleOperations(ctx context.Context, orgID string, f ConsoleOperationFilter) ([]runtimeapi.ConsoleOperation, error)

ConsoleOperations lists the organisation's connector operations, newest first.

func (*Store) ConsoleOrganization 

func (s *Store) ConsoleOrganization(ctx context.Context, orgID string) (*runtimeapi.ConsoleOrganization, error)

ConsoleOrganization returns an organisation's committed configuration with its connections' last verification results.

func (*Store) ConsoleOrganizationExecutions 

func (s *Store) ConsoleOrganizationExecutions(ctx context.Context, orgID, state string, limit int) ([]runtimeapi.ConsoleExecution, error)

ConsoleOrganizationExecutions pages the organisation's runs, newest first, optionally only those in the given state (running, completed, failed, interrupted).

func (*Store) ConsoleOrganizations 

func (s *Store) ConsoleOrganizations(ctx context.Context) ([]runtimeapi.ConsoleOrganization, error)

ConsoleOrganizations lists the active organisations' configuration.

func (*Store) ConsoleSeat 

func (s *Store) ConsoleSeat(ctx context.Context, seatID string, runs int) (*runtimeapi.ConsoleSeatDetail, error)

ConsoleSeat returns an active seat's configuration, live state, recovery state, memory metadata and recent runs.

func (*Store) ConsoleSeatStatuses 

func (s *Store) ConsoleSeatStatuses(ctx context.Context, orgID string, keys []string) ([]runtimeapi.ConsoleSeatStatus, error)

ConsoleSeatStatuses reports the live state of the organisation's active seats, optionally restricted to the given keys, ordered by key.

func (*Store) ConsoleWork added in v0.2.0 

func (s *Store) ConsoleWork(ctx context.Context, orgID string, limit int) ([]runtimeapi.ConsoleWorkItem, error)

ConsoleWork lists the organisation's work items in shared (non-personal) stores with their publication state. Only the work item schema fields are returned: this is coordination state agents chose to share, unlike notes, whose bodies the console never shows.

func (*Store) Conversation 

func (s *Store) Conversation(ctx context.Context, orgID, id string) (kind, binding string, participants []string, err error)

Conversation returns a conversation's kind, binding and participants.

func (*Store) ConversationMessages 

func (s *Store) ConversationMessages(ctx context.Context, orgID, convID string, offset, limit int) ([]Message, error)

ConversationMessages pages a conversation's messages, oldest first.

func (*Store) Conversations 

func (s *Store) Conversations(ctx context.Context, orgID, seatID string, offset, limit int) ([]ConversationSummary, error)

Conversations lists conversations the seat participates in, most recent first.

func (*Store) CreateProbe 

func (s *Store) CreateProbe(ctx context.Context, orgID, seatID string) (string, error)

CreateProbe queues a synthetic probe message for the seat; the probe id is the message id. A retiring seat is not probed.

func (*Store) CreateSchedule 

func (s *Store) CreateSchedule(ctx context.Context, f Fence, orgID, schedule string, next time.Time, note string) (*Schedule, error)

CreateSchedule registers a wake schedule for the seat itself.

func (*Store) CreateStructured added in v0.2.0 

func (s *Store) CreateStructured(ctx context.Context, f Fence, orgID, storeID, authorID, kind, dir, prefix string,
	render func(n int64) (data json.RawMessage, body string, err error)) (*Record, error)

CreateStructured creates a structured record (kind work) at "<dir>/<prefix><n>.md", allocating n from the store's sequence. render receives n and returns the record's data and rendered body.

func (*Store) DeleteOrganization 

func (s *Store) DeleteOrganization(ctx context.Context, orgID string, purge bool) error

DeleteOrganization removes an organisation's runtime. With purge=false (retain) the organisation is marked deleted and every seat retired so all capabilities are revoked while durable data stays recoverable (A19). With purge=true every row of the organisation is deleted.

func (*Store) DueRetirements added in v0.3.0 

func (s *Store) DueRetirements(ctx context.Context, limit int) ([]DueRetirement, error)

DueRetirements lists retiring seats whose grace period ended, or that have handled their retirement notice and have no queued delivery or running execution left.

func (*Store) EnqueuePublications added in v0.2.0 

func (s *Store) EnqueuePublications(ctx context.Context, orgID string, storeIDs []string, connection, kind string) error

EnqueuePublications makes sure every live work item in the stores has a publication row for connection.

func (*Store) ExpireSending 

func (s *Store) ExpireSending(ctx context.Context) (int, error)

ExpireSending marks outbound messages whose claim lapsed without a recorded outcome as unknown.

func (*Store) ExpireStaleOperations 

func (s *Store) ExpireStaleOperations(ctx context.Context, olderThan time.Duration) (int64, error)

ExpireStaleOperations marks operations left pending by a crashed attempt as unknown: the side effect may have happened, so they are never replayed.

func (*Store) FenceSeat 

func (s *Store) FenceSeat(ctx context.Context, orgID, seatID string, expected int64) (*runtimeapi.LeaseResponse, error)

FenceSeat expires the seat's lease and increments its generation so the previous holder's calls are rejected. The controller calls it only after confirming the previous Pod is gone (contracts.md).

func (*Store) FinishOperation 

func (s *Store) FinishOperation(ctx context.Context, id, status, receipt string, result json.RawMessage, errText string, attempts int) error

FinishOperation records an operation's outcome. Its result is recorded regardless of later fencing: it describes what happened externally.

func (*Store) FinishRetirement added in v0.3.0 

func (s *Store) FinishRetirement(ctx context.Context, seatID string, release WorkRelease) (*Retirement, error)

FinishRetirement retires a retiring seat in one transaction: running work is interrupted, the lease fenced, schedules stopped, undelivered messages returned to their senders, owned unfinished work items released through release, personal stores retired, and a summary sent to the organisation's representatives. It returns nil when the seat is not retiring any more.

func (*Store) FireDue 

func (s *Store) FireDue(ctx context.Context, now time.Time, limit int, next NextFunc) (int, error)

FireDue fires up to limit due schedules of active seats. Each fire inserts a schedule-origin message and delivery, deduplicated on (schedule, fire time) so a repeated fire never queues a second wake. It returns the number of wakes queued.

func (*Store) Gauges 

func (s *Store) Gauges(ctx context.Context) (*Gauges, error)

Gauges computes the gauge snapshot across active organisations.

func (*Store) Handoff 

func (s *Store) Handoff(ctx context.Context, seatID string) (*runtimeapi.Handoff, error)

Handoff returns the seat's portable handoff, or nil.

func (*Store) History 

func (s *Store) History(ctx context.Context, orgID, recordID string, storeIDs []string, offset, limit int) ([]RevisionInfo, error)

History pages a record's revisions, newest first.

func (*Store) IngestHuman 

func (s *Store) IngestHuman(ctx context.Context, in HumanMessage) (*Ingested, error)

IngestHuman persists an inbound human message and its delivery. Events are deduplicated on (organisation, connection, external event id) (A09).

func (*Store) LeaseNext 

func (s *Store) LeaseNext(ctx context.Context, f Fence) (*runtimeapi.InboxDelivery, error)

LeaseNext atomically leases the seat's oldest eligible delivery. Within a conversation deliveries are handed out in order, and a conversation with a leased delivery is skipped. The execution row records the configuration revision and lease generation the turn runs under (A14). It returns nil when nothing is eligible.

func (*Store) List added in v0.2.0 

func (s *Store) List(ctx context.Context, orgID string, storeIDs []string, prefix string, order ListOrder, desc bool, offset, limit int) ([]Entry, error)

List pages the live records of the authorised stores whose path starts with prefix.

func (*Store) LiteralCandidates added in v0.2.0 

func (s *Store) LiteralCandidates(ctx context.Context, orgID string, storeIDs []string, prefix string, queries []string,
	caseSensitive, all bool, offset, limit int) ([]Record, error)

LiteralCandidates returns live records under prefix whose body contains any (all=false) or all (all=true) of the literal substrings, newest first. The caller finds the matching lines.

func (*Store) Message 

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

Message loads a message of the organisation.

func (*Store) Migrate 

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

Migrate applies the embedded goose migrations.

func (*Store) Operation 

func (s *Store) Operation(ctx context.Context, orgID, seatID, id string) (*runtimeapi.Operation, error)

Operation loads one of the seat's operations.

func (*Store) Organization 

func (s *Store) Organization(ctx context.Context, orgID string) (*Organization, error)

Organization loads a non-deleted organisation.

func (*Store) PendingDeliveries 

func (s *Store) PendingDeliveries(ctx context.Context, seatID string) (int, error)

PendingDeliveries counts the seat's undelivered (pending or leased) messages.

func (*Store) Ping 

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

Ping checks that the database is reachable (readiness).

func (*Store) ProbeStatus 

func (s *Store) ProbeStatus(ctx context.Context, orgID, seatID, probeID string) (*runtimeapi.ProbeResponse, error)

ProbeStatus reports a probe's progress from its delivery and execution. It passes when the runner acknowledged it as completed and no reported check failed.

func (*Store) Publications added in v0.2.0 

func (s *Store) Publications(ctx context.Context, orgID string, limit int) ([]Publication, error)

Publications lists an organisation's publications, most recently changed first.

func (*Store) Record 

func (s *Store) Record(ctx context.Context, orgID, recordID string, storeIDs []string) (*Record, error)

Record reads a record from one of the authorised stores.

func (*Store) RecordByPath added in v0.2.0 

func (s *Store) RecordByPath(ctx context.Context, orgID, storeID, path string) (*Record, error)

RecordByPath reads the live record at path in an authorised store.

func (*Store) RecordConnectionCheck 

func (s *Store) RecordConnectionCheck(ctx context.Context, orgID, connection string, res runtimeapi.CheckResult, cred runtimeapi.CredentialStatus) error

RecordConnectionCheck stores a verification result and the connection's credential refresh state for readiness (§5.4). Nothing secret is stored.

func (*Store) RecordRevision 

func (s *Store) RecordRevision(ctx context.Context, orgID, recordID string, revision int, storeIDs []string) (*Record, error)

RecordRevision reads a past revision of a record in an authorised store.

func (*Store) ReleaseLease 

func (s *Store) ReleaseLease(ctx context.Context, seatID string, gen int64) error

ReleaseLease gives up the lease at generation gen and records Stopped.

func (*Store) RenewLease 

func (s *Store) RenewLease(ctx context.Context, seatID, podUID string, gen int64) (*runtimeapi.LeaseResponse, error)

RenewLease extends the lease held by podUID at generation gen.

func (*Store) ReplyToHuman 

func (s *Store) ReplyToHuman(ctx context.Context, f Fence, in HumanReply) (*Sent, error)

ReplyToHuman writes the outbound message and its outbox row in one fenced transaction; the dispatcher delivers it (§9.2).

func (*Store) ReportState 

func (s *Store) ReportState(ctx context.Context, f Fence, req runtimeapi.StateRequest) error

ReportState records the runner's execution state and adopted revision.

func (*Store) RequeueExpired 

func (s *Store) RequeueExpired(ctx context.Context) (int, error)

RequeueExpired returns deliveries whose 15-minute lease expired to the queue with backoff and dead-letters exhausted ones. It returns the number of deliveries dead-lettered.

func (*Store) ResolveOutbox 

func (s *Store) ResolveOutbox(ctx context.Context, it OutboxItem, state, receipt, errText string, retryIn time.Duration) error

ResolveOutbox records the outcome of a send. state is one of the outbox states; for OutboxPending the row is retried after retryIn.

func (*Store) ResolvePublication added in v0.2.0 

func (s *Store) ResolvePublication(ctx context.Context, it PublicationItem, state string, publishedRevision int, errText string, retryIn time.Duration) error

ResolvePublication records the outcome of an attempt. A published revision resets the attempt count; retryIn schedules the next attempt.

func (*Store) RetiringManifests added in v0.3.0 

func (s *Store) RetiringManifests(ctx context.Context, orgID string) (map[string]compile.SeatManifest, error)

RetiringManifests returns the committed manifests of the organisation's retiring seats, by key: they keep their access while winding down.

func (*Store) ReviseStructured added in v0.2.0 

func (s *Store) ReviseStructured(ctx context.Context, f Fence, orgID, recordID, authorID, kind string, expected int, storeIDs []string,
	change func(data json.RawMessage) (json.RawMessage, string, error)) (*Record, error)

ReviseStructured changes a structured record of the given kind through change, which receives the current data and returns the new data and rendered body, or an error to refuse. A stale expected revision is a *RevisionConflict; nothing is overwritten.

func (*Store) SaveCheckpoint 

func (s *Store) SaveCheckpoint(ctx context.Context, f Fence, cp runtimeapi.Checkpoint) error

SaveCheckpoint stores the seat's harness checkpoint (ADR-0005).

func (*Store) SaveHandoff 

func (s *Store) SaveHandoff(ctx context.Context, f Fence, h runtimeapi.Handoff) error

SaveHandoff replaces the seat's portable handoff (§8.4).

func (*Store) Schedules 

func (s *Store) Schedules(ctx context.Context, seatID string) ([]Schedule, error)

Schedules lists the seat's active schedules.

func (*Store) Search 

func (s *Store) Search(ctx context.Context, orgID string, storeIDs []string, query, prefix string, tags []string, offset, limit int) ([]SearchHit, error)

Search runs a Postgres full-text query (websearch syntax) over the authorised stores, optionally under a path prefix. An empty query lists the most recently updated records.

func (*Store) Seat 

func (s *Store) Seat(ctx context.Context, seatID string) (*Seat, error)

Seat loads an active seat and its committed manifest. Retired seats and seats of deleted organisations are not found, which revokes their access.

func (*Store) SeatByKey 

func (s *Store) SeatByKey(ctx context.Context, orgID, key string) (*Seat, error)

SeatByKey loads an active seat by organisation and key.

func (*Store) SeatByServiceAccount 

func (s *Store) SeatByServiceAccount(ctx context.Context, namespace, serviceAccount string) (string, error)

SeatByServiceAccount maps an authenticated ServiceAccount to its active seat.

func (*Store) SeatOrganization 

func (s *Store) SeatOrganization(ctx context.Context, seatID string) (string, error)

SeatOrganization returns the organisation of any seat, active or retired.

func (*Store) SeatRuntimes 

func (s *Store) SeatRuntimes(ctx context.Context, orgID string, keys []string) (map[string]runtimeapi.SeatRuntime, error)

SeatRuntimes reports the runtime view of the organisation's active seats, retiring ones included, optionally restricted to the given keys.

func (*Store) SendSeatMessage 

func (s *Store) SendSeatMessage(ctx context.Context, f Fence, in SeatMessage) (*Sent, error)

SendSeatMessage persists an internal message and its delivery in one fenced transaction. A full recipient inbox is rejected (ADR-0006), and a retiring or retired recipient with ErrRecipientRetiring.

func (*Store) SetPublicationExternal added in v0.2.0 

func (s *Store) SetPublicationExternal(ctx context.Context, it PublicationItem, externalID, url string) error

SetPublicationExternal records the external object a publication created, before anything else, so it is never created twice.

func (*Store) StructuredRecords added in v0.2.0 

func (s *Store) StructuredRecords(ctx context.Context, orgID, kind string, storeIDs []string, field, value string, offset, limit int) ([]Record, error)

StructuredRecords lists live structured records of kind in the authorised stores, optionally only those whose data field equals value, newest first.

func (*Store) SyncOrganization 

func (s *Store) SyncOrganization(ctx context.Context, in SyncInput) (*runtimeapi.SyncResponse, error)

SyncOrganization idempotently upserts the organisation, its seats and its memory stores in one transaction (§6.1). Seats missing from the manifest start retiring (retire.go): they take no new messages and get a bounded wind-down before they are retired with their data retained. A seat added back while retiring carries on. A seat's policy revision is bumped when its authority changes; because tools read the committed seat manifest on every call, the new policy is in force when this returns (A13).

func (*Store) TakePassive added in v0.2.0 

func (s *Store) TakePassive(ctx context.Context, f Fence, executionID string, limit int) ([]runtimeapi.Envelope, error)

TakePassive hands over the seat's queued passive messages outside a turn start (messages.inbox).

func (*Store) UnknownOperations 

func (s *Store) UnknownOperations(ctx context.Context, seatID string, limit int) ([]runtimeapi.Operation, error)

UnknownOperations lists the seat's operations whose outcome is unknown, for agent-led reconciliation (§10.3).

func (*Store) WriteFile added in v0.2.0 

func (s *Store) WriteFile(ctx context.Context, f Fence, in FileWrite) (rec *Record, created bool, err error)

WriteFile creates the note at Path, or replaces its text when it exists. Tags and source refs are kept when nil. A structured record at the path is never overwritten (ErrStructured).

type SyncInput 

type SyncInput struct {
	Namespace string
	Key       string
	SourceUID string
	Manifest  *compile.Manifest
	// RetirementGrace is how long a seat removed from the manifest may wind
	// down before it is retired; zero retires it at the next retirement pass.
	RetirementGrace time.Duration
}

SyncInput is a resolved organisation to establish identities for.

type WorkRelease added in v0.3.0 

type WorkRelease func(data json.RawMessage) (json.RawMessage, string, error)

WorkRelease rewrites a work item's data for release from a retired owner, returning the new data and rendered body.

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.