Documentation
¶
Index ¶
- func DrainMode() bool
- func JobParent(ctx context.Context) context.Context
- func WithJobParent(ctx, parent context.Context) context.Context
- type AgentHitsMaintenanceWorker
- type EntityVersionWorker
- type PartitionWorker
- type PruneWorker
- type PublishEventSink
- type PublishWorker
- type PublishedEntity
- type RetentionPurgeWorker
- type Scheduler
- func (s *Scheduler) Schedule(id string, every time.Duration, fn func(context.Context) error) error
- func (s *Scheduler) ScheduleJitter(id string, every, jitter, firstDelay time.Duration, ...) error
- func (s *Scheduler) SetAgentHitsMaintenanceWorker(worker *AgentHitsMaintenanceWorker)
- func (s *Scheduler) SetAgentHitsPartitionWorker(worker *PartitionWorker)
- func (s *Scheduler) SetAgentRangesRefreshWorker(worker *agents.VerifyWorker)
- func (s *Scheduler) SetAgentVerifyWorker(worker *agents.VerifyWorker)
- func (s *Scheduler) SetEntityVersionInterval(d time.Duration)
- func (s *Scheduler) SetEntityVersionWorker(worker *EntityVersionWorker)
- func (s *Scheduler) SetPruneWorker(worker *PruneWorker)
- func (s *Scheduler) SetPublishWorker(worker *PublishWorker)
- func (s *Scheduler) SetRetentionPurgeWorker(worker *RetentionPurgeWorker)
- func (s *Scheduler) Start(ctx context.Context)
- func (s *Scheduler) Stop()
- func (s *Scheduler) Unschedule(prefix string)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DrainMode ¶
func DrainMode() bool
DrainMode reports whether the WORKER_DRAIN env var instructs the scheduler to skip starting background worker loops on this process. Used during destructive DB migrations (e.g. Path B multi-tenancy strip) to prevent the worker fleet from racing the DDL by writing to tables mid-ALTER.
Accepted truthy values: "1", "true", "yes", "on" (case-insensitive). Anything else (including unset) returns false. Read once at Start(); to toggle, set the WORKER_DRAIN env var in the deployment's env file and restart the API process.
Types ¶
type AgentHitsMaintenanceWorker ¶
type AgentHitsMaintenanceWorker struct {
// contains filtered or unexported fields
}
AgentHitsMaintenanceWorker truncates stored client IPs after 30 days, in bounded batches so a backlog never holds a long lock.
type EntityVersionWorker ¶
type EntityVersionWorker struct {
// contains filtered or unexported fields
}
EntityVersionWorker coalesces identical-consecutive entity versions and keeps only the newest N per entity, so hatamoto_entity_versions stays bounded.
func NewEntityVersionWorker ¶
func NewEntityVersionWorker(repo entity.EntityRepository, keep int, logger *slog.Logger) *EntityVersionWorker
type PartitionWorker ¶
type PartitionWorker struct {
// contains filtered or unexported fields
}
PartitionWorker keeps a monthly RANGE-partitioned table's partitions ahead of the write boundary and drops aged ones. The parent table carries PARTITIONED indexes, so CREATE TABLE ... PARTITION OF auto-creates the matching child indexes: no manual index DDL needed. parent is always a package constant, never user input; it is also what keeps the DROP guard (nameRe) honest.
func NewAgentHitsPartitionWorker ¶
func NewAgentHitsPartitionWorker(db *database.DB, aheadMonths, retentionMonths int, logger *slog.Logger) *PartitionWorker
NewAgentHitsPartitionWorker maintains hatamoto_agent_hits. retentionMonths=3 keeps at least 90 days of hits (month granularity: up to ~120 days).
func (*PartitionWorker) EnsureCurrentAndNext ¶
func (w *PartitionWorker) EnsureCurrentAndNext(ctx context.Context)
EnsureCurrentAndNext synchronously creates the current and next month's partitions and warns when the DEFAULT partition holds rows. It is meant for server startup, independent of the scheduler (HATAMOTO_WORKER=off is the default), so hits never land in DEFAULT where retention cannot drop them. Errors are logged, never returned: startup must not depend on it.
type PruneWorker ¶
type PruneWorker struct {
// contains filtered or unexported fields
}
PruneWorker performs batched DELETEs on high-churn tables so they don't grow unbounded. Each delete is done in bounded LIMIT batches (via ctid) so a large backlog never takes a long lock.
func NewPruneWorker ¶
type PublishEventSink ¶
type PublishEventSink interface {
EntityPublished(ctx context.Context, id uuid.UUID, schemaSlug, slug, name string, version int)
}
PublishEventSink is told about each entity the worker promotes so the plugin host can fire entity.published. Defined here (primitives only) to keep this package free of entity/pluginhost imports.
type PublishWorker ¶
type PublishWorker struct {
// contains filtered or unexported fields
}
PublishWorker promotes scheduled entities to published once their published_at falls due.
Why a status rather than a date filter: the public repository (internal/entity/public_repository.go) selects on status = 'published' across eight query sites with no date guard. Adding "AND published_at <= NOW()" to each would work until someone adds a ninth and leaks a post early. A distinct 'scheduled' status is excluded by every one of those queries for free, and this worker is the single place the transition happens.
The UPDATE deliberately does not touch published_at: the scheduled instant is the publication date. entity.UpdateStatus only stamps time.Now() when published_at is NULL or still in the future (a post published early by hand), so it keeps a past date, but the worker stays a separate, explicit path.
func NewPublishWorker ¶
func NewPublishWorker(db *database.DB, logger *slog.Logger) *PublishWorker
NewPublishWorker constructs the worker.
func (*PublishWorker) Run ¶
func (w *PublishWorker) Run(ctx context.Context) ([]PublishedEntity, error)
Run executes one tick: promote every scheduled entity whose published_at has passed. Returns the rows promoted.
Each promoted entity goes through the events sink, which clears the page cache (the server's sink does so synchronously) so the page and listings are live as soon as Run returns, whatever HATAMOTO_PAGE_CACHE_TTL is.
func (*PublishWorker) SetEvents ¶
func (w *PublishWorker) SetEvents(s PublishEventSink)
SetEvents installs the sink; nil disables notification.
type PublishedEntity ¶
PublishedEntity identifies one row this worker promoted.
type RetentionPurgeWorker ¶
type RetentionPurgeWorker struct {
// contains filtered or unexported fields
}
RetentionPurgeWorker deletes expired server-side sessions once per day. session.Store.PurgeExpired already existed but had no scheduled caller.
func NewRetentionPurgeWorker ¶
func NewRetentionPurgeWorker(sessions *session.Store, logger *slog.Logger) *RetentionPurgeWorker
NewRetentionPurgeWorker constructs the worker. sessions may be nil (purge skipped).
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler manages background job scheduling.
The scheduler is a small core-only host for core maintenance jobs (publishing, retention, pruning, agent-hit upkeep). Plugin workloads run in the plugin via the plugin.Starter contract, not here.
func NewScheduler ¶
NewScheduler creates a new scheduler instance.
func (*Scheduler) Schedule ¶
Schedule runs fn every `every` (at least a minute; the first run is one period from now) under id, replacing a job with the same id. Jobs added before Start begin with it; in drain mode they never run. It is how plugin cron registrations reach the scheduler.
func (*Scheduler) ScheduleJitter ¶
func (s *Scheduler) ScheduleJitter(id string, every, jitter, firstDelay time.Duration, fn func(context.Context) error) error
ScheduleJitter is Schedule with a first run after firstDelay and a wait between runs of every plus a uniform offset in [-jitter, +jitter]. A job with the same id is replaced. The 10 minute cap per run still applies.
func (*Scheduler) SetAgentHitsMaintenanceWorker ¶
func (s *Scheduler) SetAgentHitsMaintenanceWorker(worker *AgentHitsMaintenanceWorker)
SetAgentHitsMaintenanceWorker registers the 30-day client-IP truncation job.
func (*Scheduler) SetAgentHitsPartitionWorker ¶
func (s *Scheduler) SetAgentHitsPartitionWorker(worker *PartitionWorker)
SetAgentHitsPartitionWorker registers the hatamoto_agent_hits partition job.
func (*Scheduler) SetAgentRangesRefreshWorker ¶
func (s *Scheduler) SetAgentRangesRefreshWorker(worker *agents.VerifyWorker)
SetAgentRangesRefreshWorker registers the daily operator IP range download.
func (*Scheduler) SetAgentVerifyWorker ¶
func (s *Scheduler) SetAgentVerifyWorker(worker *agents.VerifyWorker)
SetAgentVerifyWorker registers the 5-minute pending-hit verification job.
func (*Scheduler) SetEntityVersionInterval ¶
SetEntityVersionInterval makes the entity-version job run every d (after a minute) instead of daily. The public demo uses it so old versions cannot pile up between two resets.
func (*Scheduler) SetEntityVersionWorker ¶
func (s *Scheduler) SetEntityVersionWorker(worker *EntityVersionWorker)
SetEntityVersionWorker registers the entity-version coalesce/retention job.
func (*Scheduler) SetPruneWorker ¶
func (s *Scheduler) SetPruneWorker(worker *PruneWorker)
SetPruneWorker registers the batched high-churn-table prune job.
func (*Scheduler) SetPublishWorker ¶
func (s *Scheduler) SetPublishWorker(worker *PublishWorker)
SetPublishWorker registers the scheduled-publish worker (optional). When set, the scheduler promotes due `scheduled` entities once a minute.
func (*Scheduler) SetRetentionPurgeWorker ¶
func (s *Scheduler) SetRetentionPurgeWorker(worker *RetentionPurgeWorker)
SetRetentionPurgeWorker registers the daily session + remember-token purge.
func (*Scheduler) Start ¶
Start begins the scheduler loops.
When WORKER_DRAIN=true in the environment, all worker loops are skipped. The scheduler comes up but does nothing. This is the safety hook used before running destructive DB migrations: set the env, restart the API process, wait for any in-flight work (claimed before drain) to finish, verify the queue is empty, then run the migration.
func (*Scheduler) Unschedule ¶
Unschedule stops every job added with Schedule whose id starts with prefix.
Source Files