package workflow import ( "context" "encoding/json" "errors" "log/slog" "strings" "sync" "github.com/nomadcode/nomadcode-core/internal/storage" ) // externalRefLocks provides per-key mutual exclusion for external ref operations. // It is safe for concurrent use. In-process only: multi-instance deployments // still require a distributed lock (e.g. Postgres advisory lock) to prevent // cross-instance duplicate side effects. type externalRefLocks struct { mu sync.Mutex entries map[string]*refEntry } type refEntry struct { mu sync.Mutex holders int } func (r *externalRefLocks) acquire(key string) *refEntry { r.mu.Lock() if r.entries == nil { r.entries = make(map[string]*refEntry) } e, ok := r.entries[key] if !ok { e = &refEntry{} r.entries[key] = e } e.holders++ r.mu.Unlock() e.mu.Lock() return e } func (r *externalRefLocks) release(key string, e *refEntry) { e.mu.Unlock() r.mu.Lock() e.holders-- if e.holders == 0 { delete(r.entries, key) } r.mu.Unlock() } var ( ErrInvalidTaskInput = errors.New("invalid task input") ErrTaskCannotBeEnqueued = errors.New("task cannot be enqueued in current status") ErrInvalidTaskTransition = errors.New("invalid task transition") ) type TaskEnqueuer interface { EnqueueTask(ctx context.Context, taskID string) error } type Service struct { store *storage.Store enqueuer TaskEnqueuer logger *slog.Logger lifecycle *Lifecycle extRefLocks externalRefLocks } func NewService(store *storage.Store, enqueuer TaskEnqueuer, logger *slog.Logger) *Service { return &Service{ store: store, enqueuer: enqueuer, logger: logger, lifecycle: NewLifecycle(store, logger), } } func (s *Service) WithExternalRefLock(ctx context.Context, provider, id string, fn func(ctx context.Context) (storage.Task, error)) (storage.Task, error) { key := provider + "\x00" + id e := s.extRefLocks.acquire(key) defer s.extRefLocks.release(key, e) return fn(ctx) } func (s *Service) CreateTask(ctx context.Context, input CreateTaskInput) (storage.Task, error) { title := strings.TrimSpace(input.Title) source := strings.TrimSpace(input.Source) if title == "" || source == "" { return storage.Task{}, ErrInvalidTaskInput } payload := input.Payload if len(payload) == 0 || string(payload) == "null" { payload = json.RawMessage(`{}`) } if !json.Valid(payload) { return storage.Task{}, ErrInvalidTaskInput } metadata, err := NormalizeTaskMetadata(input.Metadata) if err != nil { return storage.Task{}, err } external, err := NormalizeExternalRef(input.External) if err != nil { return storage.Task{}, err } return s.store.CreateTask(ctx, storage.CreateTaskInput{ Title: title, Source: source, Payload: payload, Metadata: metadata, ExternalProvider: external.Provider, ExternalID: external.ID, ExternalURL: external.URL, ExternalMetadata: external.Metadata, }) } func (s *Service) GetTask(ctx context.Context, id string) (storage.Task, error) { return s.store.GetTask(ctx, id) } func (s *Service) GetTaskByExternalRef(ctx context.Context, provider, id string) (storage.Task, error) { provider = strings.TrimSpace(provider) id = strings.TrimSpace(id) if provider == "" || id == "" { return storage.Task{}, ErrInvalidTaskInput } return s.store.GetTaskByExternalRef(ctx, provider, id) } type NormalizedExternalRef struct { Provider *string ID *string URL *string Metadata json.RawMessage } func NormalizeExternalRef(input *ExternalRefInput) (NormalizedExternalRef, error) { if input == nil { return NormalizedExternalRef{Metadata: json.RawMessage(`{}`)}, nil } provider := strings.TrimSpace(input.Provider) id := strings.TrimSpace(input.ID) externalURL := strings.TrimSpace(input.URL) metadata := input.Metadata if len(metadata) == 0 || string(metadata) == "null" { metadata = json.RawMessage(`{}`) } if !json.Valid(metadata) { return NormalizedExternalRef{}, ErrInvalidTaskInput } if provider == "" { return NormalizedExternalRef{}, ErrInvalidTaskInput } return NormalizedExternalRef{ Provider: optionalString(provider), ID: optionalString(id), URL: optionalString(externalURL), Metadata: metadata, }, nil } func optionalString(value string) *string { if value == "" { return nil } return &value } func NormalizeTaskMetadata(metadata json.RawMessage) (json.RawMessage, error) { if len(metadata) == 0 || string(metadata) == "null" { return json.RawMessage(`{}`), nil } if !json.Valid(metadata) { return nil, ErrInvalidTaskInput } return metadata, nil } func (s *Service) ListTasks(ctx context.Context, limit int32) ([]storage.Task, error) { if limit <= 0 { limit = 20 } if limit > 100 { limit = 100 } return s.store.ListTasks(ctx, limit) } func (s *Service) UpdateTaskMetadata(ctx context.Context, id string, metadata json.RawMessage) (storage.Task, error) { normalized, err := NormalizeTaskMetadata(metadata) if err != nil { return storage.Task{}, err } return s.store.UpdateMetadata(ctx, id, normalized) } func (s *Service) EnqueueTask(ctx context.Context, id string) (storage.Task, error) { queuedTask, err := s.lifecycle.QueueTask(ctx, id) if err != nil { return storage.Task{}, err } if s.enqueuer == nil { err := errors.New("task enqueuer is not configured") if _, failErr := s.lifecycle.FailTask(ctx, id, err.Error()); failErr != nil && s.logger != nil { s.logger.Error("failed to mark task enqueue error", "task_id", id, "error", failErr) } return queuedTask, err } if err := s.enqueuer.EnqueueTask(ctx, id); err != nil { if _, failErr := s.lifecycle.FailTask(ctx, id, err.Error()); failErr != nil && s.logger != nil { s.logger.Error("failed to mark task enqueue error", "task_id", id, "error", failErr) } return queuedTask, err } return queuedTask, nil } func canEnqueue(status string) bool { return canEnqueueStatus(TaskStatus(status)) } func (s *Service) SetEnqueuer(enqueuer TaskEnqueuer) { s.enqueuer = enqueuer } func (s *Service) CompleteTaskByExternalRef(ctx context.Context, provider, id string, result json.RawMessage) error { _, err := s.WithExternalRefLock(ctx, provider, id, func(ctx context.Context) (storage.Task, error) { task, err := s.lifecycle.store.GetTaskByExternalRef(ctx, provider, id) if err != nil { return storage.Task{}, err } if task.Status == string(StatusCompleted) { return task, nil } return s.lifecycle.CompleteTask(ctx, task.ID, result) }) return err }