131 lines
4.2 KiB
Go
131 lines
4.2 KiB
Go
// Package scheduler — system_builtins.go
|
|
//
|
|
// v0.28.6: Built-in system functions registered at startup.
|
|
// These mirror the background goroutines in main.go but run as visible,
|
|
// scheduled, auditable tasks. The goroutines remain as fallback until
|
|
// system tasks are validated in production.
|
|
package scheduler
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
"time"
|
|
|
|
"git.gobha.me/xcaliber/chat-switchboard/database"
|
|
"git.gobha.me/xcaliber/chat-switchboard/store"
|
|
"git.gobha.me/xcaliber/chat-switchboard/taskutil"
|
|
)
|
|
|
|
// RegisterBuiltins registers all built-in system functions.
|
|
// Called once from main.go at startup.
|
|
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
|
|
}
|
|
// Default: expire sessions older than 7 days
|
|
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) {
|
|
// Default: 48 hours idle → stale
|
|
cutoff := time.Now().UTC().Add(-48 * time.Hour)
|
|
res, err := database.DB.ExecContext(ctx, database.Q(`
|
|
UPDATE channels
|
|
SET workflow_status = 'stale'
|
|
WHERE type = 'workflow'
|
|
AND workflow_status = 'active'
|
|
AND last_activity_at < $1
|
|
`), cutoff)
|
|
if err != nil {
|
|
return "", fmt.Errorf("staleness sweep failed: %w", err)
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
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) {
|
|
if database.IsSQLite() {
|
|
return "skipped: retention enforcement requires PostgreSQL (JSON operators)", nil
|
|
}
|
|
cutoff := time.Now().UTC().Add(-48 * time.Hour)
|
|
res, err := database.DB.ExecContext(ctx, database.Q(`
|
|
DELETE FROM channels
|
|
WHERE type = 'workflow'
|
|
AND workflow_status IN ('completed', 'archived')
|
|
AND workflow_id IS NOT NULL
|
|
AND last_activity_at < $1
|
|
AND workflow_id IN (
|
|
SELECT id FROM workflows
|
|
WHERE retention IS NOT NULL
|
|
AND retention->>'mode' = 'delete'
|
|
AND (retention->>'delete_after_days')::int > 0
|
|
AND channels.last_activity_at < now() - ((retention->>'delete_after_days')::int || ' days')::interval
|
|
)
|
|
`), cutoff)
|
|
if err != nil {
|
|
return "", fmt.Errorf("retention enforcement failed: %w", err)
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
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) {
|
|
cutoff := time.Now().UTC().Add(-7 * 24 * time.Hour)
|
|
res, err := database.DB.ExecContext(ctx, database.Q(`
|
|
DELETE FROM health_windows WHERE window_start < $1
|
|
`), cutoff)
|
|
if err != nil {
|
|
return "", fmt.Errorf("health prune failed: %w", err)
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
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
|
|
}
|