package jobs import ( "context" "errors" "evobgp/internal/store" ) // FindIdempotentContext looks up a retained logical operation without inserting. func (r *Registry) FindIdempotentContext(ctx context.Context, tenant, key string) (*Job, error) { if r.queue != nil { c, err := r.queue.FindJobByIdempotencyKey(ctx, tenant, key) if err != nil { return nil, err } j := jobFromRecord(c) j.requestHash = c.RequestHash return j, nil } r.mu.RLock() defer r.mu.RUnlock() j := r.byIdempo[idempoKey{tenant: tenant, key: key}] if j == nil { return nil, store.ErrJobNotFound } return j, nil } // ReplayContext validates an immutable request before expensive live resolution. func (r *Registry) ReplayContext(ctx context.Context, in store.JobEnqueue) (*Job, bool, error) { if in.IdempotencyKey == nil || *in.IdempotencyKey == "" { return nil, false, nil } j, err := r.FindIdempotentContext(ctx, in.TenantID, *in.IdempotencyKey) if errors.Is(err, store.ErrJobNotFound) { return nil, false, nil } if err != nil { return nil, false, err } h, err := store.JobRequestHash(in) if err != nil { return nil, false, err } if j.requestHash != h { return nil, false, store.ErrJobConflict } return j, true, nil } // GetContext reads a job from the shared store when durability is enabled. func (r *Registry) GetContext(ctx context.Context, tenant, id string) (*Job, error) { if r.queue == nil { return r.Get(tenant, id) } c, err := r.queue.GetJob(ctx, tenant, id) if errors.Is(err, store.ErrJobNotFound) { return nil, ErrNotFound } if err != nil { return nil, err } return jobFromRecord(c), nil } // ListContext returns a bounded tenant page from the authoritative queue. func (r *Registry) ListContext(ctx context.Context, tenant, status, kind, cursor string, limit int) ([]*Job, string, bool, error) { if r.queue == nil { j, n, m := r.List(tenant, status, kind, cursor, limit) return j, n, m, nil } c, next, more, err := r.queue.ListJobs(ctx, tenant, status, kind, cursor, limit) if err != nil { return nil, "", false, err } out := make([]*Job, 0, len(c)) for _, v := range c { out = append(out, jobFromRecord(v)) } return out, next, more, nil } // RequestCancelContext persists a cancel request visible to any worker replica. func (r *Registry) RequestCancelContext(ctx context.Context, tenant, id string) (*Job, error) { if r.queue == nil { return r.RequestCancel(tenant, id) } c, err := r.queue.CancelJob(ctx, tenant, id) if errors.Is(err, store.ErrJobNotFound) { return nil, ErrNotFound } if err != nil { return nil, err } r.mu.RLock() local := r.byID[id] r.mu.RUnlock() if local != nil { local.RequestCancel() } return jobFromRecord(c), nil } // RetryContext schedules an explicit retry of a failed shared operation. func (r *Registry) RetryContext(ctx context.Context, tenant, id string) (*Job, error) { if r.queue == nil { return nil, store.ErrJobConflict } c, err := r.queue.RetryJob(ctx, tenant, id) if err != nil { return nil, err } return jobFromRecord(c), nil }