func (s *QSvc) Modify(ctx context.Context, req *pb.ModifyRequest) (*pb.ModifyResponse, error) {
	if err := s.Authorize(ctx, s.modifyAuthz(ctx, req)); err != nil {
		return nil, err  
	}
	claimant, err := uuid.Parse(req.ClaimantId)
	if err != nil {
		return nil, codeErrorf(codes.InvalidArgument, "failed to parse claimant ID: %w", err)
	}
	modArgs := []entroq.ModifyArg{
		entroq.ModifyAs(claimant),
	}
	for _, insert := range req.Inserts {
		var (
			id  uuid.UUID
			err error
		)
		if insert.Id != "" {
			if id, err = uuid.Parse(insert.Id); err != nil {
				return nil, codeErrorf(codes.InvalidArgument, "failed to parse explicit insertion ID: %w", err)
			}
		}
		modArgs = append(modArgs,
			entroq.InsertingInto(insert.Queue,
				entroq.WithArrivalTime(fromMS(insert.AtMs)),
				entroq.WithValue(insert.Value),
				entroq.WithAttempt(insert.Attempt),
				entroq.WithErr(insert.Err),
				entroq.WithID(id)))
	}
	for _, change := range req.Changes {
		id, err := uuid.Parse(change.GetOldId().Id)
		if err != nil {
			return nil, codeErrorf(codes.InvalidArgument, "failed to parse change id: %w", err)
		}
		t := &entroq.Task{
			ID:        id,
			Version:   change.GetOldId().Version,
			Claimant:  claimant,
			Queue:     change.GetNewData().Queue,
			Value:     change.GetNewData().Value,
			At:        fromMS(change.GetNewData().AtMs),
			Attempt:   change.GetNewData().Attempt,
			Err:       change.GetNewData().Err,
			FromQueue: change.GetOldId().Queue,
		}
		modArgs = append(modArgs, entroq.Changing(t))
	}
	for _, del := range req.Deletes {
		id, err := uuid.Parse(del.Id)
		if err != nil {
			return nil, codeErrorf(codes.InvalidArgument, "failed to parse deletion id: %w", err)
		}
		modArgs = append(modArgs, entroq.Deleting(id, del.Version, entroq.WithIDQueue(del.Queue)))
	}
	for _, dep := range req.Depends {
		id, err := uuid.Parse(dep.Id)
		if err != nil {
			return nil, codeErrorf(codes.InvalidArgument, "failed to parse depency id: %w", err)
		}
		modArgs = append(modArgs, entroq.DependingOn(id, dep.Version, entroq.WithIDQueue(dep.Queue)))
	}
	inserted, changed, err := s.impl.Modify(ctx, modArgs...)
	if err != nil {
		if depErr, ok := entroq.AsDependency(err); ok {
			tmap := map[pb.ActionType][]*entroq.TaskID{
				pb.ActionType_INSERT: depErr.Inserts,
				pb.ActionType_DEPEND: depErr.Depends,
				pb.ActionType_DELETE: depErr.Deletes,
				pb.ActionType_CHANGE: depErr.Changes,
				pb.ActionType_CLAIM:  depErr.Claims,
			}
			details := []proto.Message{&pb.ModifyDep{
				Type: pb.ActionType_DETAIL,
				Msg:  depErr.Message,
			}}
			for dtype, dvals := range tmap {
				for _, tid := range dvals {
					details = append(details, &pb.ModifyDep{
						Type: dtype,
						Id:   &pb.TaskID{Id: tid.ID.String(), Version: tid.Version, Queue: tid.Queue},
					})
				}
			}
			stat, sErr := status.New(codes.NotFound, "modification dependency error").WithDetails(details...)
			if sErr != nil {
				return nil, codeErrorf(codes.NotFound, "dependency failed, and failed to add details %v: %w", err, sErr)
			}
			return nil, stat.Err()
		}
		return nil, autoCodeErrorf("modification failed: %w", err)
	}
	resp := new(pb.ModifyResponse)
	for _, task := range inserted {
		resp.Inserted = append(resp.Inserted, protoFromTask(task))
	}
	for _, task := range changed {
		resp.Changed = append(resp.Changed, protoFromTask(task))
	}
	return resp, nil
}
Modify attempts to make the specified modification from the given
ModifyRequest. If all goes well, it returns a ModifyResponse. If the
modification fails due to a dependency error (one of the specified tasks was
not present), the gRPC status mechanism is invoked to return a status with
the details slice containing *pb.ModifyDep values. These could be used to
reconstruct an entroq.DependencyError, or directly to find out which IDs
caused the dependency failure. Code UNKNOWN is returned on other errors.
package qsvc
import (
	"context"
	"errors"
	"fmt"
	"log"
	"time"
	"github.com/golang/protobuf/proto"
	"github.com/google/uuid"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/shiblon/entroq"
	"github.com/shiblon/entroq/pkg/authz"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	pb "github.com/shiblon/entroq/proto"
)
const (
	MetricNS = "entroq"
)
var (
	metricQueueSize = promauto.NewGaugeVec(
		prometheus.GaugeOpts{
			Namespace: MetricNS,
			Subsystem: "queue",
			Name:      "size",
			Help:      "Number of tasks in named queue.",
		},
		[]string{"name", "type"},
	)
)
type QSvc struct {
	pb.UnimplementedEntroQServer
	impl *entroq.EntroQ
	metricCancel    context.CancelFunc
	metricNamespace string
	metricInterval  time.Duration
	authzHeader string
	az          authz.Authorizer
}
type Option func(*QSvc)
func WithMetricInterval(d time.Duration) Option {
	return func(s *QSvc) {
		if d < time.Minute {
			d = time.Minute
		}
		s.metricInterval = d
	}
}
func WithAuthorizationHeader(h string) Option {
	return func(s *QSvc) {
		s.authzHeader = h
	}
}
func WithAuthorizer(az authz.Authorizer) Option {
	return func(s *QSvc) {
		s.az = az
	}
}
func New(ctx context.Context, opener entroq.BackendOpener, opts ...Option) (*QSvc, error) {
	impl, err := entroq.New(ctx, opener)
	if err != nil {
		return nil, fmt.Errorf("qsvc backend client: %w", err)
	}
	mctx, mcancel := context.WithCancel(context.Background())
	svc := &QSvc{
		impl:           impl,
		metricCancel:   mcancel,
		metricInterval: time.Minute,
		authzHeader:    "authorization",
	}
	for _, o := range opts {
		o(svc)
	}
	go func() {
		for {
			if err := svc.RefreshMetrics(mctx); err != nil {
				log.Printf("Error refreshing metrics: %v", err)
			}
			select {
			case <-mctx.Done():
				return
			case <-time.After(svc.metricInterval):
			}
		}
	}()
	return svc, nil
}
func (s *QSvc) Close() error {
	s.metricCancel()  
	if err := s.impl.Close(); err != nil {
		return fmt.Errorf("qsvc close: %w", err)
	}
	return nil
}
func (s *QSvc) Authorize(ctx context.Context, req *authz.Request) error {
	if s.az == nil {
		return nil
	}
	if err := s.az.Authorize(ctx, req); err != nil {
		var details []proto.Message
		authzErr := new(authz.AuthzError)
		if !errors.As(err, &authzErr) {
			return status.New(codes.PermissionDenied, fmt.Sprintf("unknown authz error: %v", err)).Err()
		}
		for _, msg := range authzErr.Errors {
			details = append(details, &pb.AuthzDep{
				Actions: []pb.ActionType{pb.ActionType_DETAIL},
				Msg:     msg,
			})
		}
		for _, q := range authzErr.Failed {
			var actions []pb.ActionType
			for _, a := range q.Actions {
				switch a {
				case "READ":
					actions = append(actions, pb.ActionType_READ)
				case "INSERT":
					actions = append(actions, pb.ActionType_INSERT)
				case "CLAIM":
					actions = append(actions, pb.ActionType_CLAIM)
				case "DELETE":
					actions = append(actions, pb.ActionType_DELETE)
				case "CHANGE":
					actions = append(actions, pb.ActionType_CHANGE)
				default:
					details = append(details, &pb.AuthzDep{
						Actions: []pb.ActionType{pb.ActionType_DETAIL},
						Exact:   fmt.Sprintf("WARNING: Unknown action %q", a),
					})
				}
			}
			details = append(details, &pb.AuthzDep{
				Actions: actions,
				Exact:   q.Exact,
				Prefix:  q.Prefix,
			})
		}
		stat, sErr := status.New(codes.PermissionDenied, "queue action denied").WithDetails(details...)
		if sErr != nil {
			return status.New(codes.PermissionDenied, fmt.Sprintf("queue action denied, unable to add details to %v: %v", err, sErr)).Err()
		}
		return stat.Err()
	}
	return nil
}
func (s *QSvc) RefreshMetrics(ctx context.Context) error {
	stats, err := s.impl.QueueStats(ctx)
	if err != nil {
		return fmt.Errorf("refresh metrics: %w", err)
	}
	metricQueueSize.Reset()
	for name, stat := range stats {
		metricQueueSize.WithLabelValues(name, "total").Set(float64(stat.Size))
		metricQueueSize.WithLabelValues(name, "claimed").Set(float64(stat.Claimed))
		metricQueueSize.WithLabelValues(name, "available").Set(float64(stat.Available))
		metricQueueSize.WithLabelValues(name, "maxClaims").Set(float64(stat.MaxClaims))
	}
	return nil
}
func fromMS(ms int64) time.Time {
	return time.Unix(0, ms*int64(time.Millisecond))
}
func toMS(t time.Time) int64 {
	return t.Truncate(time.Millisecond).UnixNano() / 1000000
}
func protoFromTask(t *entroq.Task) *pb.Task {
	return &pb.Task{
		Queue:      t.Queue,
		Id:         t.ID.String(),
		Version:    t.Version,
		AtMs:       toMS(t.At),
		ClaimantId: t.Claimant.String(),
		Claims:     t.Claims,
		Value:      t.Value,
		CreatedMs:  toMS(t.Created),
		ModifiedMs: toMS(t.Modified),
		Attempt:    t.Attempt,
		Err:        t.Err,
	}
}
func autoCodeErrorf(format string, vals ...interface{}) error {
	err := fmt.Errorf(format, vals...)
	if entroq.IsTimeout(err) {
		return status.New(codes.DeadlineExceeded, err.Error()).Err()
	}
	if entroq.IsCanceled(err) {
		return status.New(codes.Canceled, err.Error()).Err()
	}
	return err
}
func codeErrorf(code codes.Code, format string, vals ...interface{}) error {
	return status.New(code, fmt.Errorf(format, vals...).Error()).Err()
}
func (s *QSvc) authzToken(ctx context.Context) string {
	md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return ""
	}
	vals := md[s.authzHeader]
	if len(vals) == 0 {
		return ""
	}
	return vals[0]
}
func (s *QSvc) newAuthzRequest(ctx context.Context) *authz.Request {
	return &authz.Request{
		Authz: authz.NewHeaderAuthorization(s.authzToken(ctx)),
	}
}
func (s *QSvc) claimAuthz(ctx context.Context, req *pb.ClaimRequest) *authz.Request {
	authReq := s.newAuthzRequest(ctx)
	for _, q := range req.GetQueues() {
		authReq.Queues = append(authReq.Queues, &authz.Queue{
			Exact:   q,
			Actions: []authz.Action{authz.Claim},
		})
	}
	return authReq
}
func (s *QSvc) tasksAuthz(ctx context.Context, req *pb.TasksRequest) *authz.Request {
	authReq := s.newAuthzRequest(ctx)
	authReq.Queues = append(authReq.Queues, &authz.Queue{
		Exact:   req.Queue,
		Actions: []authz.Action{authz.Read},
	})
	return authReq
}
func (s *QSvc) modifyAuthz(ctx context.Context, req *pb.ModifyRequest) *authz.Request {
	authReq := s.newAuthzRequest(ctx)
	for _, ins := range req.Inserts {
		authReq.Queues = append(authReq.Queues, &authz.Queue{
			Exact:   ins.Queue,
			Actions: []authz.Action{authz.Insert},
		})
	}
	for _, chg := range req.Changes {
		oldQueue, newQueue := chg.GetOldId().Queue, chg.GetNewData().Queue
		if oldQueue == newQueue {
			authReq.Queues = append(authReq.Queues, &authz.Queue{
				Exact:   newQueue,
				Actions: []authz.Action{authz.Change},
			})
		} else {
			authReq.Queues = append(authReq.Queues,
				&authz.Queue{
					Exact:   oldQueue,
					Actions: []authz.Action{authz.Delete},
				},
				&authz.Queue{
					Exact:   newQueue,
					Actions: []authz.Action{authz.Insert},
				},
			)
		}
	}
	for _, del := range req.Deletes {
		authReq.Queues = append(authReq.Queues, &authz.Queue{
			Exact:   del.Queue,
			Actions: []authz.Action{authz.Delete},
		})
	}
	for _, dep := range req.Depends {
		authReq.Queues = append(authReq.Queues, &authz.Queue{
			Exact:   dep.Queue,
			Actions: []authz.Action{authz.Read},
		})
	}
	return authReq
}
func (s *QSvc) Claim(ctx context.Context, req *pb.ClaimRequest) (*pb.ClaimResponse, error) {
	if err := s.Authorize(ctx, s.claimAuthz(ctx, req)); err != nil {
		return nil, err  
	}
	claimant, err := uuid.Parse(req.ClaimantId)
	if err != nil {
		return nil, codeErrorf(codes.InvalidArgument, "failed to parse claimant ID: %w", err)
	}
	duration := time.Duration(req.DurationMs) * time.Millisecond
	pollTime := time.Duration(0)
	if req.PollMs > 0 {
		pollTime = time.Duration(req.PollMs) * time.Millisecond
	}
	task, err := s.impl.Claim(ctx,
		entroq.From(req.Queues...),
		entroq.ClaimFor(duration),
		entroq.ClaimAs(claimant),
		entroq.ClaimPollTime(pollTime))
	if err != nil {
		return nil, autoCodeErrorf("qsvc claim: %w", err)
	}
	if task == nil {
		return new(pb.ClaimResponse), nil
	}
	return &pb.ClaimResponse{Task: protoFromTask(task)}, nil
}
func (s *QSvc) TryClaim(ctx context.Context, req *pb.ClaimRequest) (*pb.ClaimResponse, error) {
	if err := s.Authorize(ctx, s.claimAuthz(ctx, req)); err != nil {
		return nil, err  
	}
	claimant, err := uuid.Parse(req.ClaimantId)
	if err != nil {
		return nil, codeErrorf(codes.InvalidArgument, "failed to parse claimant ID: %w", err)
	}
	duration := time.Duration(req.DurationMs) * time.Millisecond
	task, err := s.impl.TryClaim(ctx,
		entroq.From(req.Queues...),
		entroq.ClaimFor(duration),
		entroq.ClaimAs(claimant))
	if err != nil {
		return nil, autoCodeErrorf("try claim: %w", err)
	}
	if task == nil {
		return new(pb.ClaimResponse), nil
	}
	return &pb.ClaimResponse{Task: protoFromTask(task)}, nil
}