Documentation
¶
Overview ¶
Package watch implements project Watches over DALGO transactional state.
Index ¶
- Constants
- Variables
- func ValidateDefinition(definition Definition) error
- type Actor
- type AuditEvent
- type Claim
- type CloudRetainedDocument
- type CloudRetainedInventory
- type CloudRetainedInventoryLimits
- type CloudRetainedSource
- type CloudRetainedWrite
- type CloudRetainedWriteAdmission
- type CloudRetainedWritePlan
- type Completion
- type Config
- type Definition
- type DefinitionRepository
- type Delivery
- type Event
- type FirestoreOptions
- type Notification
- type NotificationPolicy
- type Observation
- type RecipientAuthority
- type Run
- type Runner
- type RunnerIdentity
- type Schedule
- type Scope
- type Status
- type Store
- func (s *Store) AdvanceTelegramPoll(ctx context.Context, id RunnerIdentity, fence, offset int64) error
- func (s *Store) Audit(ctx context.Context, a Actor, watchID string) ([]AuditEvent, error)
- func (s *Store) BeginTelegramLink(ctx context.Context, a Actor, nonceHash string, expiresAt time.Time) error
- func (s *Store) ClaimDue(ctx context.Context, identity RunnerIdentity) (*Claim, error)
- func (s *Store) ClaimNotifications(ctx context.Context) (*Delivery, error)
- func (s *Store) ClaimTelegramNotification(ctx context.Context, id RunnerIdentity, pollFence int64) (*Delivery, error)
- func (s *Store) ClaimTelegramPoll(ctx context.Context, id RunnerIdentity) (*TelegramPollLease, error)
- func (s *Store) Close() error
- func (s *Store) CompleteTelegramLinkByRunner(ctx context.Context, id RunnerIdentity, pollFence int64, ...) error
- func (s *Store) DeferTelegramPoll(ctx context.Context, id RunnerIdentity, fence int64, retryAfterSeconds int) error
- func (s *Store) DeleteDefinition(ctx context.Context, a Actor, id, operationID string, expectedVersion int64) error
- func (s *Store) DisconnectTelegram(ctx context.Context, a Actor) error
- func (s *Store) FinishNotification(ctx context.Context, id string, fence int64, failureCode string) error
- func (s *Store) FinishRun(ctx context.Context, identity RunnerIdentity, c Completion) (Run, error)
- func (s *Store) FinishTelegramNotification(ctx context.Context, identity RunnerIdentity, pollFence int64, id string, ...) error
- func (s *Store) GetClaim(ctx context.Context, id RunnerIdentity, runID string, fence int64) (Claim, error)
- func (s *Store) GetDefinition(ctx context.Context, a Actor, id string) (Definition, error)
- func (s *Store) GetEvent(ctx context.Context, a Actor, id string) (Event, error)
- func (s *Store) GetSubscription(ctx context.Context, a Actor, watchID string) (Subscription, error)
- func (s *Store) Health(ctx context.Context) error
- func (s *Store) Heartbeat(ctx context.Context, identity RunnerIdentity, id string, fence int64) error
- func (s *Store) Inbox(ctx context.Context, a Actor) ([]Notification, error)
- func (s *Store) ListDefinitions(ctx context.Context, a Actor) ([]Definition, error)
- func (s *Store) ListDefinitionsInTransaction(ctx context.Context, tx dal.ReadTransaction, actor Actor) ([]Definition, error)
- func (s *Store) ListDeliveries(ctx context.Context, a Actor, watchID string) ([]Delivery, error)
- func (s *Store) ListEventPage(ctx context.Context, a Actor, watchID, before string, limit int) ([]Event, string, error)
- func (s *Store) ListEvents(ctx context.Context, a Actor, watchID string) ([]Event, error)
- func (s *Store) ListRuns(ctx context.Context, a Actor, watchID string) ([]Run, error)
- func (s *Store) MarkRead(ctx context.Context, a Actor, id string) error
- func (s *Store) NextWakeup(ctx context.Context) (time.Time, bool, error)
- func (s *Store) Prune(ctx context.Context, maxRows int) error
- func (s *Store) PutDefinition(ctx context.Context, a Actor, d Definition, operationID string, ...) (Definition, error)
- func (s *Store) ReadDefinitionInTransaction(ctx context.Context, tx dal.ReadTransaction, actor Actor, id string) (Definition, error)
- func (s *Store) ReadPutDefinitionOperationInTransaction(ctx context.Context, tx dal.ReadTransaction, actor Actor, requested Definition, ...) (Definition, bool, error)
- func (s *Store) Reconcile(ctx context.Context) error
- func (s *Store) ReconcileScope(ctx context.Context, scope Scope, maxWatches int) (bool, error)
- func (s *Store) RegisterRunner(ctx context.Context, identity RunnerIdentity) error
- func (s *Store) RenewTelegramPoll(ctx context.Context, id RunnerIdentity, fence int64) (*TelegramPollLease, error)
- func (s *Store) ResetBaseline(ctx context.Context, a Actor, id, operationID string, expectedVersion int64) (Definition, error)
- func (s *Store) RetainedProjectInventoryInTransaction(ctx context.Context, tx dal.ReadTransaction, ...) (CloudRetainedInventory, error)
- func (s *Store) RetryNotification(ctx context.Context, a Actor, id string) error
- func (s *Store) RevalidateSources(ctx context.Context, scope Scope, maxWatches int, after string) (string, error)
- func (s *Store) SetLifecycle(ctx context.Context, a Actor, id, lifecycle, operationID string, ...) (Definition, error)
- func (s *Store) Status(ctx context.Context, a Actor) (Status, error)
- func (s *Store) Subscribe(ctx context.Context, a Actor, sub Subscription) error
- func (s *Store) TelegramProviderStatus(ctx context.Context, a Actor) (TelegramProviderStatus, error)
- func (s *Store) TelegramStatus(ctx context.Context, a Actor) (bool, error)
- func (s *Store) WithClaim(ctx context.Context, id RunnerIdentity, runID string, fence int64, ...) error
- type Subscription
- type TelegramPollLease
- type TelegramProviderStatus
- type WorkEvent
Constants ¶
const CloudDefinitionCollection = definitions
Stable source labels for trusted typed inventory/admission adapters.
const CloudRunCollection = runs
const MaxEventPageSize = 1000
Variables ¶
var ( ErrForbidden = errors.New("watch access forbidden") ErrNotFound = errors.New("watch resource not found") ErrConflict = errors.New("watch version conflict") ErrStaleLease = errors.New("watch lease expired or superseded") ErrInvalid = errors.New("invalid watch request") )
var ErrCapacity = errors.New("watch operation exceeds its bounded storage budget")
Functions ¶
func ValidateDefinition ¶ added in v0.119.0
func ValidateDefinition(definition Definition) error
ValidateDefinition validates the native Watch definition without reading or writing a catalog. It lets editors review embedded proposals using the same structural rules as Store.PutDefinition.
This does not authorize saving or execution. PutDefinition must still enforce current scope, source, ownership, subscriptions, capacity and configured limits.
Types ¶
type AuditEvent ¶
type Claim ¶
type Claim struct {
ClaimID string `json:"claimId"`
RunID string `json:"runId"`
Fence int64 `json:"fence"`
LeaseUntil time.Time `json:"leaseUntil"`
// Deadline is copied from the stored Run on admission. It is the original
// execution ceiling for this fenced attempt; heartbeat never extends it.
Deadline time.Time `json:"deadline,omitempty"`
Definition Definition `json:"definition"`
}
func (Claim) ExecutionCeiling ¶ added in v0.139.0
ExecutionCeiling validates a coordinator-admitted claim's original deadline and current lease, then narrows that deadline by independently authenticated original credential/enrollment ceilings. Zero or expired bounds fail closed. It authenticates nothing: a protected worker must obtain the claim and bounds from trusted admission and pin the returned ceiling once for this attempt. A later heartbeat can extend LeaseUntil, never that pinned ceiling.
type CloudRetainedDocument ¶ added in v0.128.0
type CloudRetainedDocument struct {
Collection string
ID string
Environment string
WatchID string
Owner string
Payload json.RawMessage
}
CloudRetainedDocument is an internal storage envelope, not an API response. Payload may contain private inputs and delivery details. Reference projection must decode its table's concrete schema, including historical operation pins.
type CloudRetainedInventory ¶ added in v0.128.0
type CloudRetainedInventory struct {
Namespace string
Project string
Collections []string
Documents []CloudRetainedDocument
}
type CloudRetainedInventoryLimits ¶ added in v0.128.0
type CloudRetainedInventoryLimits struct{ MaxDocuments, MaxBytes int }
type CloudRetainedSource ¶ added in v0.128.0
type CloudRetainedSource struct {
Collection, RecordID string
Owner string
Scope Scope
WatchID, RunID, EventID string
DeliveryID string
Active bool
Definition *Definition
Run *Run
Event *Event
Observation *Observation
}
CloudRetainedSource is private typed resolver input, not a lifecycle response. Scope is preserved independently for every owner/environment. The host must join Definition/Run/Event pins to its immutable source manifests and classify other project sources before it can claim a complete reference inventory. Expiry and coordinator retention cutoffs deliberately do not remove rows.
func ProjectCloudRetainedSources ¶ added in v0.128.0
func ProjectCloudRetainedSources(ctx context.Context, inventory CloudRetainedInventory) ([]CloudRetainedSource, error)
ProjectCloudRetainedSources validates the closed table schemas and their physical envelope joins. It does not infer dependencies from a notification URL, human text, inputs, opaque CheckRunRef or the current mutable Check. A legacy payload, unexpected table/action, or incomplete scan returns no rows.
type CloudRetainedWrite ¶ added in v0.128.0
type CloudRetainedWrite struct {
Before *CloudRetainedDocument
After *CloudRetainedDocument
}
CloudRetainedWrite represents a project-owned change. Nil Before is an insert; nil After is a physical removal. Both are immutable copies of the envelopes, not a reference graph or a claim that retained dependencies can be purged.
type CloudRetainedWriteAdmission ¶ added in v0.128.0
type CloudRetainedWriteAdmission func(context.Context, dal.ReadwriteTransaction, string, string, []CloudRetainedWrite) (CloudRetainedWritePlan, error)
type CloudRetainedWritePlan ¶ added in v0.128.0
CloudRetainedWriteAdmission must validate typed dependencies and read the project fence in the supplied transaction. It must not write during prepare. For a nonempty batch it must return a nonnil write-only persistence closure, with conservative row/byte costs for every additional write. These are charged against the existing transaction budget before any physical write is issued. Errors from either stage abort the entire source transaction. Domain-specific authorization and runner/recipient admission remain independently required.
type Completion ¶
type Completion struct {
SourceRevision string `json:"sourceRevision,omitempty"`
SourceFingerprint string `json:"sourceFingerprint,omitempty"`
ClaimID string `json:"claimId"`
RunID string `json:"runId"`
Fence int64 `json:"fence"`
ExecutionStatus string `json:"executionStatus"`
Outcome string `json:"outcome"`
Severity string `json:"severity,omitempty"`
DurationMS int64 `json:"durationMs"`
CheckRunRef string `json:"checkRunRef,omitempty"`
FailureCode string `json:"failureCode,omitempty"`
}
type Config ¶
type Config struct {
MaxWatchesPerUser int
MinIntervalSeconds int
TelegramDailyLimit int
TelegramMinuteLimit int
IsPaidAccount func(string) bool
EmailDailyLimit int
SMSDailyLimit int
EmailMinuteLimit int
SMSMinuteLimit int
Namespace string
Now func() time.Time
LeaseDuration time.Duration
RunnerTTL time.Duration
ExecutionTimeout time.Duration
MaxAttempts int
HistoryLimit int
}
type Definition ¶
type Definition struct {
// SourceRevision is a canonical cloud admission pin, not a client grant.
SourceRevision string `json:"sourceRevision,omitempty"`
SourceFingerprint string `json:"sourceFingerprint,omitempty"`
Scope
ID string `json:"id"`
Version int64 `json:"version"`
Title string `json:"title"`
Owner string `json:"owner"`
Visibility string `json:"visibility"`
CheckID string `json:"checkId"`
CheckVersion string `json:"checkVersion,omitempty"`
Inputs map[string]detective.InputValue `json:"inputs,omitempty"`
Schedule Schedule `json:"schedule"`
Trigger string `json:"trigger"` // becomes_true, becomes_false, changes
InitialObservation string `json:"initialObservation"` // notify, baseline
Mode string `json:"mode"` // one_shot, continuous
Lifecycle string `json:"lifecycle"` // enabled, paused, completed, deleted
Enabled bool `json:"enabled"`
Requirements []string `json:"requirements,omitempty"`
NotificationPolicy NotificationPolicy `json:"notificationPolicy"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
type DefinitionRepository ¶
type DefinitionRepository interface {
ListDefinitions(context.Context, Actor) ([]Definition, error)
GetDefinition(context.Context, Actor, string) (Definition, error)
PutDefinition(context.Context, Actor, Definition, string, int64) (Definition, error)
}
DefinitionRepository is durable project storage, independent of job/lease retention. This deployment supplies authenticated DALGO storage; personal definitions are never exported to a project's shared Git files.
type Delivery ¶
type Delivery struct {
TelegramLinkGeneration int64 `json:"telegramLinkGeneration,omitempty"`
TelegramBotID string `json:"telegramBotId,omitempty"`
TelegramChatID string `json:"telegramChatId,omitempty"`
AccountID string `json:"accountId,omitempty"`
Phone string `json:"phone,omitempty"`
QuotaReserved bool `json:"quotaReserved"`
Notification
Channel string `json:"channel"`
Email string `json:"email,omitempty"`
State string `json:"state"`
Attempts int `json:"attempts"`
Fence int64 `json:"fence"`
LeaseUntil time.Time `json:"leaseUntil,omitempty"`
RetryAt time.Time `json:"retryAt,omitempty"`
FailureCode string `json:"failureCode,omitempty"`
}
type Event ¶ added in v0.116.0
type Event struct {
Scope
ID string `json:"id"`
WatchID string `json:"watchId"`
RunID string `json:"runId"`
DefinitionVersion int64 `json:"definitionVersion"`
Title string `json:"title"`
Kind string `json:"kind"` // trigger or recovery
Previous string `json:"previous"` // true, false, never
Outcome string `json:"outcome"` // pass or fail
OccurredAt time.Time `json:"occurredAt"`
CheckID string `json:"checkId"`
CheckVersion string `json:"checkVersion"`
SourceFingerprint string `json:"sourceFingerprint"`
CheckRunRef string `json:"checkRunRef,omitempty"`
Inputs map[string]detective.InputValue `json:"inputs,omitempty"`
Owner string `json:"owner"`
Visibility string `json:"visibility"`
}
Event is the immutable accepted business transition, independent of notification subscriptions and cooldown. Inputs are definition snapshots, never runner rows.
type FirestoreOptions ¶ added in v0.115.0
type FirestoreOptions struct {
ProjectKey *record.Key
// StorageRoot is a separately protected server-only project container. It
// must not be a descendant of a project subtree readable by Space members.
StorageRoot *record.Key
AccountRoot *record.Key
// AccountNamespace is a server-fixed deployment quota namespace. Empty uses Config.Namespace.
AccountNamespace string
ProjectID string
Config Config
Authorize func(context.Context, dal.ReadTransaction) error
AdmitDefinition func(context.Context, dal.ReadTransaction, Definition, string) (string, error)
AdmitRunner func(context.Context, dal.ReadTransaction, Definition, RunnerIdentity) (bool, error)
AdmitRecipient func(context.Context, dal.ReadTransaction, Definition, string) (RecipientAuthority, error)
// AdmitRetainedWrite prepares reference admission after the complete buffered
// command, before any physical write. Its returned closure may only write and
// must commit the shared project fence in this same transaction. Installing
// this hook does not establish that other project writers participate.
AdmitRetainedWrite CloudRetainedWriteAdmission
// DefinitionChanged stages exactly one <=1KiB deterministic metadata event.
// It runs after every read, must not read/publish/network, and shares commit.
DefinitionChanged func(context.Context, dal.ReadwriteTransaction, Definition) error
// WorkReady stages one deterministic metadata wakeup for new pending work.
// It shares DefinitionChanged's no-read/no-network/<=1KiB contract.
WorkReady func(context.Context, dal.ReadwriteTransaction, WorkEvent) error
}
FirestoreOptions binds one Store to an existing trusted project and store/Space namespace. AccountRoot must be shared by projects in that same namespace so creating another project cannot evade account/channel quotas. Both keys and all callbacks are constructed by the authenticated server, never request JSON.
type Notification ¶
type Notification struct {
EventID string `json:"eventId,omitempty"`
Scope
ID string `json:"id"`
WatchID string `json:"watchId"`
RunID string `json:"runId"`
UserID string `json:"userId"`
Title string `json:"title"`
Event string `json:"event"`
Link string `json:"link"`
CreatedAt time.Time `json:"createdAt"`
ReadAt time.Time `json:"readAt,omitempty"`
}
type NotificationPolicy ¶
type Observation ¶
type Observation struct {
LastEventID string `json:"lastEventId,omitempty"`
SourceFingerprint string `json:"sourceFingerprint,omitempty"`
WatchID string `json:"watchId"`
Value string `json:"value"` // true, false, unknown, never
LastKnown string `json:"lastKnown"`
LastRunID string `json:"lastRunId,omitempty"`
LastObservedAt time.Time `json:"lastObservedAt,omitempty"`
LastNotifiedAt time.Time `json:"lastNotifiedAt,omitempty"`
NextEvaluation time.Time `json:"nextEvaluation"`
Availability string `json:"availability"`
}
type RecipientAuthority ¶ added in v0.115.0
RecipientAuthority is fresh server-owned identity and entitlement evidence. Paid is never populated from a browser request or an unverified token claim.
type Run ¶
type Run struct {
EventID string `json:"eventId,omitempty"`
Scope
ID string `json:"id"`
WatchID string `json:"watchId"`
DefinitionVersion int64 `json:"definitionVersion"`
DueAt time.Time `json:"dueAt"`
State string `json:"state"` // due, running, completed, retry, failed, cancelled
RunnerID string `json:"runnerId,omitempty"`
Fence int64 `json:"fence"`
Attempts int `json:"attempts"`
Deadline time.Time `json:"deadline,omitempty"`
CompletionHash string `json:"completionHash,omitempty"`
LeaseUntil time.Time `json:"leaseUntil,omitempty"`
RetryAt time.Time `json:"retryAt,omitempty"`
ExecutionStatus string `json:"executionStatus,omitempty"`
Outcome string `json:"outcome,omitempty"`
Severity string `json:"severity,omitempty"`
DurationMS int64 `json:"durationMs,omitempty"`
CheckRunRef string `json:"checkRunRef,omitempty"`
FailureCode string `json:"failureCode,omitempty"`
FinishedAt time.Time `json:"finishedAt,omitempty"`
}
type RunnerIdentity ¶
type RunnerIdentity struct {
TelegramBotID string
TelegramBotUsername string
ID string
Scope Scope
Capabilities []string
}
RunnerIdentity contains only grants verified by the coordinator transport.
type Status ¶
type Status struct {
Coordinator string `json:"coordinator"`
Runners []Runner `json:"runners"`
Observations []Observation `json:"observations"`
}
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
func NewFirestoreStore ¶ added in v0.115.0
func NewFirestoreStore(db dal.DB, options FirestoreOptions) (*Store, error)
NewFirestoreStore reuses the pinned DALGO Firestore adapter. It neither creates a database nor owns its shared client; Store.Close leaves that client open. Authorization executes inside the same transaction as protected reads/writes. Callback errors abort the transaction, and callbacks may run more than once.
func NewSQLiteStore ¶
NewSQLiteStore accepts an absolute local file path or sqlite:///absolute/path. Storage configuration belongs to daemon deployment. Arbitrary driver DSNs are refused so a caller cannot override transaction locking or durability pragmas.
func (*Store) AdvanceTelegramPoll ¶
func (*Store) BeginTelegramLink ¶
func (s *Store) BeginTelegramLink(ctx context.Context, a Actor, nonceHash string, expiresAt time.Time) error
BeginTelegramLink binds a short-lived connection request to the authenticated user, project scope and one registered bot. No browser supplies a chat ID.
func (*Store) ClaimNotifications ¶
ClaimNotifications is an internal coordinator worker API. HTTP must never expose it to browser users or runners: it contains delivery-only addresses.
func (*Store) ClaimTelegramNotification ¶
func (*Store) ClaimTelegramPoll ¶
func (s *Store) ClaimTelegramPoll(ctx context.Context, id RunnerIdentity) (*TelegramPollLease, error)
func (*Store) CompleteTelegramLinkByRunner ¶
func (s *Store) CompleteTelegramLinkByRunner(ctx context.Context, id RunnerIdentity, pollFence int64, nonceHash, chatID string) error
CompleteTelegramLinkByRunner accepts proof only from the fenced provider poller whose operator grant owns this bot and scope. Replayed messages for the same chat are idempotent; another chat never steals a consumed connection.
func (*Store) DeferTelegramPoll ¶
func (s *Store) DeferTelegramPoll(ctx context.Context, id RunnerIdentity, fence int64, retryAfterSeconds int) error
DeferTelegramPoll persists Telegram's bot-wide flood wait and fences all old requests, so failover to a second runner cannot bypass the provider cooldown.
func (*Store) DeleteDefinition ¶
func (*Store) DisconnectTelegram ¶
func (*Store) FinishNotification ¶
func (*Store) FinishRun ¶
func (s *Store) FinishRun(ctx context.Context, identity RunnerIdentity, c Completion) (Run, error)
func (*Store) FinishTelegramNotification ¶
func (*Store) GetClaim ¶ added in v0.115.0
func (s *Store) GetClaim(ctx context.Context, id RunnerIdentity, runID string, fence int64) (Claim, error)
GetClaim validates the current lease, definition version, runner eligibility, owner/source admission and baseline in one transaction before a bundle read.
func (*Store) GetDefinition ¶
func (*Store) GetEvent ¶ added in v0.116.0
GetEvent checks current visibility in the same transaction as reading evidence. Historical personal inputs never gain access through a stale authorization read.
func (*Store) GetSubscription ¶
func (*Store) Health ¶
Health reads persisted storage, so an unavailable file is not reported healthy.
func (*Store) ListDefinitions ¶
func (*Store) ListDefinitionsInTransaction ¶ added in v0.133.0
func (s *Store) ListDefinitionsInTransaction(ctx context.Context, tx dal.ReadTransaction, actor Actor) ([]Definition, error)
ListDefinitionsInTransaction lists the actor's currently visible definitions in a trusted host transaction from this Store's database. It applies exactly the same scope and personal/shared visibility as ListDefinitions, without reading runs or other owners' private retained history.
func (*Store) ListDeliveries ¶
func (*Store) ListEventPage ¶ added in v0.116.0
func (s *Store) ListEventPage(ctx context.Context, a Actor, watchID, before string, limit int) ([]Event, string, error)
ListEventPage bounds database work and returns newest first. The opaque cursor is an event ID in this exact Watch and scope; timestamp ties use storage IDs. Events are durable evidence and deliberately excluded from runtime pruning.
func (*Store) ListEvents ¶ added in v0.116.0
func (*Store) NextWakeup ¶ added in v0.115.0
NextWakeup returns the next bounded scheduling/delivery wakeup. No work means a scale-to-zero queue handler can stop scheduling tasks. Telegram polling is runner-owned and does not keep a coordinator task chain alive.
func (*Store) Prune ¶ added in v0.115.0
Prune is bounded maintenance for terminal history and expired helper records. Pending jobs/deliveries are never selected, so an accumulated history cannot wedge claim/complete operations or silently discard unpublished work.
func (*Store) PutDefinition ¶
func (s *Store) PutDefinition(ctx context.Context, a Actor, d Definition, operationID string, expectedVersion int64) (Definition, error)
func (*Store) ReadDefinitionInTransaction ¶ added in v0.120.0
func (s *Store) ReadDefinitionInTransaction(ctx context.Context, tx dal.ReadTransaction, actor Actor, id string) (Definition, error)
ReadDefinitionInTransaction reads a visible native Watch in a trusted host's existing transaction. The host must supply a transaction from this Store's underlying database, before physical writes, and a server-authenticated actor. DALGO transactions expose no portable database identity witness; this method must never be used with a transaction selected by a client.
Current Store authorization and scope/visibility rules are rechecked in the supplied snapshot. Callers must separately validate current source admission before reusing a definition for an operation. This method never writes.
func (*Store) ReadPutDefinitionOperationInTransaction ¶ added in v0.121.0
func (s *Store) ReadPutDefinitionOperationInTransaction(ctx context.Context, tx dal.ReadTransaction, actor Actor, requested Definition, operationID string, expectedVersion int64) (Definition, bool, error)
ReadPutDefinitionOperationInTransaction confirms an accepted save without executing it. A missing receipt does not prove a save never occurred. It requires the original pre-save definition, actor, operation ID and expected version. Missing receipts return found=false and never grant permission to retry a save that has lost human approval.
As for ReadDefinitionInTransaction, the trusted host supplies this Store's underlying transaction before physical writes and a server-authenticated actor. Current project authorization and current/saved visibility are rechecked. Callers separately recheck source admission before using the receipt for another operation. This method never writes or runs an admission provider.
func (*Store) Reconcile ¶
Reconcile generates at most one due execution per Watch in one transaction. Missed intervals coalesce to the first evaluation after service resumes; a job keeps its stable ID over retries and restarts. Claiming is a separate operation.
func (*Store) ReconcileScope ¶ added in v0.115.0
ReconcileScope handles at most maxWatches due observations; queue wakeups can repeat it. Each Watch gets its own atomic transaction, avoiding a namespace snapshot and one transaction's budget growing with unrelated project history.
func (*Store) RegisterRunner ¶
func (s *Store) RegisterRunner(ctx context.Context, identity RunnerIdentity) error
func (*Store) RenewTelegramPoll ¶
func (s *Store) RenewTelegramPoll(ctx context.Context, id RunnerIdentity, fence int64) (*TelegramPollLease, error)
func (*Store) ResetBaseline ¶
func (s *Store) ResetBaseline(ctx context.Context, a Actor, id, operationID string, expectedVersion int64) (Definition, error)
ResetBaseline explicitly acknowledges a reviewed source revision. Cosmetic edits deliberately cannot clear the accepted semantic fingerprint.
func (*Store) RetainedProjectInventoryInTransaction ¶ added in v0.128.0
func (s *Store) RetainedProjectInventoryInTransaction(ctx context.Context, tx dal.ReadTransaction, limits CloudRetainedInventoryLimits) (CloudRetainedInventory, error)
RetainedProjectInventoryInTransaction reads every owner/environment in this configured hosted coordinator, including deleted definitions, expired rows, completed runs and immutable receipts/events. It never uses visibility lanes, HistoryLimit, ready/live filters or retention cutoffs.
The authenticated server must prove tx uses the SAME database as the Store. This internal method performs fresh project authorization but is not a user content-read API: caller must redact source identities separately. It starts no transaction, writes nothing, and grants no completeness for external/native coordinators or other project sources. Unknown schemas, bounds and read/close failures must disable lifecycle mutations; no partial result is returned.
func (*Store) RetryNotification ¶
RetryNotification retries an owned failed delivery using its original event and quota reservation. Revoked subscriptions are checked again before dispatch.
func (*Store) RevalidateSources ¶ added in v0.115.0
func (s *Store) RevalidateSources(ctx context.Context, scope Scope, maxWatches int, after string) (string, error)
RevalidateSources checks at most twenty project definitions after a durable document-ID cursor. Source-change consumers persist the returned continuation before acknowledging their producer event. An empty cursor means the bounded pass is complete. Transient admission errors abort without blocking a Watch.
func (*Store) SetLifecycle ¶
func (*Store) TelegramProviderStatus ¶
func (*Store) TelegramStatus ¶
func (*Store) WithClaim ¶ added in v0.115.0
func (s *Store) WithClaim(ctx context.Context, id RunnerIdentity, runID string, fence int64, fn func(context.Context, dal.ReadTransaction, Claim) error) error
WithClaim admits and reads the immutable source bundle in the exact same authorized snapshot as the current fenced lease. Callback results are usable only when this method returns nil; callbacks must not perform external work.
type Subscription ¶
type Subscription struct {
TelegramBotID string `json:"telegramBotId,omitempty"`
AccountID string `json:"accountId,omitempty"`
Tier string `json:"tier,omitempty"`
Phone string `json:"phone,omitempty"`
WatchID string `json:"watchId"`
UserID string `json:"userId"`
Channels []string `json:"channels"`
NotifyOn []string `json:"notifyOn,omitempty"`
Email string `json:"email,omitempty"`
Unsubscribed bool `json:"unsubscribed,omitempty"`
}
type TelegramPollLease ¶
type TelegramPollLease struct {
NextAttemptAt time.Time `json:"nextAttemptAt,omitempty"`
BotID string `json:"botId"`
Fence int64 `json:"fence"`
Offset int64 `json:"offset"`
LeaseUntil time.Time `json:"leaseUntil"`
}
TelegramPollLease fences a single provider poller and its persisted update offset. Offset is advanced only after an update has been handled.
type TelegramProviderStatus ¶
type TelegramProviderStatus struct {
ReconnectRequired bool `json:"reconnectRequired,omitempty"`
ConnectionError string `json:"connectionError,omitempty"`
Connected bool `json:"connected"`
BotID string `json:"botId,omitempty"`
BotUsername string `json:"botUsername,omitempty"`
}
TelegramProviderStatus exposes only the registered public bot identity and the requesting user's verified link. Bot secrets remain on provider runners.
Source Files
¶
- claim_deadline.go
- definition_operation_read.go
- definition_read.go
- definitions.go
- events.go
- firestore.go
- firestore_admission.go
- firestore_changes.go
- firestore_claim.go
- firestore_events.go
- firestore_fence.go
- firestore_inbox.go
- firestore_operations.go
- firestore_query.go
- firestore_retention.go
- firestore_session.go
- model.go
- notifications.go
- retained_admission.go
- retained_inventory.go
- retained_sources.go
- retention.go
- scheduler.go
- storage.go
- telegram.go
- telegram_provider.go
- validate.go