toki
f762d2abd9
- Add worker storage layer with PostgreSQL/sqlc setup - Add migration files for worker backbone schema - Add job runner and built-in jobs implementation - Add Redis key definitions for worker state management - Archive socket-session-loop milestone (completed/renamed) - Update roadmap current.md and foundation-alignment phase - Add socket endpoint integration and runtime smoke tests - Add infra-check, worker-storage-check, worker-storage-gen CLI tools - Update API socket server and config - Add pubspec dependencies and client socket endpoint changes - Add config tests for API and worker services
66 lines
1.5 KiB
Go
66 lines
1.5 KiB
Go
package jobs
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sync"
|
|
)
|
|
|
|
// Handler defines the function signature for executing a job.
|
|
type Handler func(ctx context.Context, payload json.RawMessage) error
|
|
|
|
// Runner manages the registration and execution of worker job handlers.
|
|
type Runner struct {
|
|
mu sync.RWMutex
|
|
handlers map[Kind]Handler
|
|
}
|
|
|
|
// NewRunner creates a new Runner instance.
|
|
func NewRunner() *Runner {
|
|
return &Runner{
|
|
handlers: make(map[Kind]Handler),
|
|
}
|
|
}
|
|
|
|
// Register registers a handler for a specific job kind.
|
|
func (r *Runner) Register(kind Kind, handler Handler) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.handlers[kind] = handler
|
|
}
|
|
|
|
// Len returns the number of registered handlers.
|
|
func (r *Runner) Len() int {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
return len(r.handlers)
|
|
}
|
|
|
|
// Execute dispatches the job to its registered handler and executes it.
|
|
// It checks context cancellation and handles panics gracefully.
|
|
func (r *Runner) Execute(ctx context.Context, job Job) error {
|
|
r.mu.RLock()
|
|
handler, exists := r.handlers[job.Kind]
|
|
r.mu.RUnlock()
|
|
|
|
if !exists {
|
|
return fmt.Errorf("unknown job kind: %s", job.Kind)
|
|
}
|
|
|
|
// Check context before starting
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
return r.executeWithPanicRecovery(ctx, handler, job.Payload)
|
|
}
|
|
|
|
func (r *Runner) executeWithPanicRecovery(ctx context.Context, handler Handler, payload json.RawMessage) (err error) {
|
|
defer func() {
|
|
if rec := recover(); rec != nil {
|
|
err = fmt.Errorf("job panicked: %v", rec)
|
|
}
|
|
}()
|
|
return handler(ctx, payload)
|
|
}
|