package store import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "time" ) // SystemJobTenantID owns global maintenance jobs outside user tenant scope. const SystemJobTenantID = "00000000-0000-0000-0000-000000000000" // JobRequestHash identifies the immutable requested operation, excluding result metadata. func JobRequestHash(in JobEnqueue) (string, error) { meta := in.Meta if in.Request != nil { meta = in.Request } b, err := json.Marshal(struct { Kind string `json:"kind"` ModuleID *string `json:"module_id"` Meta map[string]any `json:"meta"` }{in.Kind, in.ModuleID, meta}) if err != nil { return "", err } h := sha256.Sum256(b) return hex.EncodeToString(h[:]), nil } var ( ErrJobNotFound = errors.New("store: job not found") ErrJobConflict = errors.New("store: idempotency key conflicts with another request") ErrJobQueueFull = errors.New("store: job queue is full") ErrJobLeaseLost = errors.New("store: job lease lost") ) // JobRecord is the shared, tenant-scoped job lifecycle record. type JobRecord struct { RequestHash string ID, TenantID, Kind, Status string IdempotencyKey, ModuleID *string Meta map[string]any CreatedAt time.Time StartedAt, FinishedAt *time.Time LeaseUntil *time.Time Error *string OwnerID string FencingToken int64 RequestSequence int64 Attempt int CancelRequested bool } // JobEnqueue is the immutable request used for idempotent insertion. type JobEnqueue struct { TenantID, Kind string IdempotencyKey, ModuleID *string Meta map[string]any // Request optionally supplies the original selector before dynamic resolution. Request map[string]any } // JobQueue provides durable operations; PostgreSQL is authoritative in production. type JobQueue interface { CountJobsByStatus(context.Context) (map[string]int64, error) FindJobByIdempotencyKey(context.Context, string, string) (JobRecord, error) EnqueueJob(context.Context, JobEnqueue) (JobRecord, bool, error) ClaimJobs(context.Context, string, int, time.Duration) ([]JobRecord, error) // ReapExpiredJobs marks orphaned cancellations and exhausted retries as terminal. // Production callers run it about every 15 seconds, not on every claim. ReapExpiredJobs(context.Context) error HeartbeatJob(context.Context, string, string, int64, time.Duration) (bool, error) CheckpointJob(context.Context, JobRecord) error FinishJob(context.Context, JobRecord) error ReleaseJob(context.Context, string, string, int64) error GetJob(context.Context, string, string) (JobRecord, error) ListJobs(context.Context, string, string, string, string, int) ([]JobRecord, string, bool, error) CancelJob(context.Context, string, string) (JobRecord, error) RetryJob(context.Context, string, string) (JobRecord, error) }