100 lines
3.0 KiB
Go
100 lines
3.0 KiB
Go
// Package scheduler — system_builtins.go
|
|
//
|
|
// v0.28.6: Built-in system functions registered at startup.
|
|
// v0.29.0: Raw SQL replaced with store methods.
|
|
package scheduler
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
"time"
|
|
|
|
"chat-switchboard/store"
|
|
"chat-switchboard/taskutil"
|
|
)
|
|
|
|
// RegisterBuiltins registers all built-in system functions.
|
|
func RegisterBuiltins() {
|
|
taskutil.RegisterSystemFunc("session_cleanup",
|
|
"Remove expired anonymous visitor sessions",
|
|
sessionCleanup)
|
|
|
|
taskutil.RegisterSystemFunc("staleness_check",
|
|
"Mark idle workflow instances as stale",
|
|
stalenessCheck)
|
|
|
|
taskutil.RegisterSystemFunc("retention_sweep",
|
|
"Delete completed workflow instances past retention policy",
|
|
retentionSweep)
|
|
|
|
taskutil.RegisterSystemFunc("health_prune",
|
|
"Prune provider health windows older than 7 days",
|
|
healthPrune)
|
|
}
|
|
|
|
// ── session_cleanup ─────────────────────────
|
|
|
|
func sessionCleanup(ctx context.Context, stores store.Stores) (string, error) {
|
|
if stores.Sessions == nil {
|
|
return "skipped: session store not available", nil
|
|
}
|
|
cutoff := time.Now().UTC().AddDate(0, 0, -7)
|
|
n, err := stores.Sessions.DeleteExpired(ctx, cutoff)
|
|
if err != nil {
|
|
return "", fmt.Errorf("session cleanup failed: %w", err)
|
|
}
|
|
msg := fmt.Sprintf("cleaned up %d expired sessions (cutoff: %s)", n, cutoff.Format(time.RFC3339))
|
|
if n > 0 {
|
|
log.Printf("[system_task] %s", msg)
|
|
}
|
|
return msg, nil
|
|
}
|
|
|
|
// ── staleness_check ─────────────────────────
|
|
|
|
func stalenessCheck(ctx context.Context, stores store.Stores) (string, error) {
|
|
cutoff := time.Now().UTC().Add(-48 * time.Hour)
|
|
n, err := stores.Channels.MarkStaleWorkflows(ctx, cutoff)
|
|
if err != nil {
|
|
return "", fmt.Errorf("staleness sweep failed: %w", err)
|
|
}
|
|
msg := fmt.Sprintf("marked %d idle workflow instances as stale (cutoff: %s)", n, cutoff.Format(time.RFC3339))
|
|
if n > 0 {
|
|
log.Printf("[system_task] %s", msg)
|
|
}
|
|
return msg, nil
|
|
}
|
|
|
|
// ── retention_sweep ─────────────────────────
|
|
|
|
func retentionSweep(ctx context.Context, stores store.Stores) (string, error) {
|
|
n, err := stores.Channels.EnforceWorkflowRetention(ctx)
|
|
if err != nil {
|
|
return "", fmt.Errorf("retention enforcement failed: %w", err)
|
|
}
|
|
msg := fmt.Sprintf("deleted %d expired workflow instances (retention policy)", n)
|
|
if n > 0 {
|
|
log.Printf("[system_task] %s", msg)
|
|
}
|
|
return msg, nil
|
|
}
|
|
|
|
// ── health_prune ────────────────────────────
|
|
|
|
func healthPrune(ctx context.Context, stores store.Stores) (string, error) {
|
|
if stores.Health == nil {
|
|
return "skipped: health store not available", nil
|
|
}
|
|
cutoff := time.Now().UTC().Add(-7 * 24 * time.Hour)
|
|
n, err := stores.Health.Prune(ctx, cutoff)
|
|
if err != nil {
|
|
return "", fmt.Errorf("health prune failed: %w", err)
|
|
}
|
|
msg := fmt.Sprintf("pruned %d old health windows (cutoff: %s)", n, cutoff.Format(time.RFC3339))
|
|
if n > 0 {
|
|
log.Printf("[system_task] %s", msg)
|
|
}
|
|
return msg, nil
|
|
}
|