Queues and logs

// Read hands up to n new messages to consumer. Each one enters the pending entries list under
// that consumer, with a delivery count of 1, and stays there until it is acknowledged.
func (s *Stream) Read(ctx context.Context, consumer string, n int64) ([]Msg, error) {
  res, err := s.RDB.XReadGroup(ctx, &redis.XReadGroupArgs{
    Group: s.Group, Consumer: consumer, Streams: []string{s.Key, ">"}, Count: n, Block: -1,
  }).Result()
  if errors.Is(err, redis.Nil) {
    return nil, nil
  }
  if err != nil {
    return nil, fmt.Errorf("read for %s: %w", consumer, err)
  }
  var out []Msg
  for _, st := range res {
    for _, m := range st.Messages {
      out = append(out, Msg{ID: m.ID, Job: fmt.Sprint(m.Values["job"]), Deliveries: 1})
    }
  }
  return out, nil
}

// Ack removes a message from the pending entries list. The entry stays in the stream.
func (s *Stream) Ack(ctx context.Context, id string) error {
  if err := s.RDB.XAck(ctx, s.Key, s.Group, id).Err(); err != nil {
    return fmt.Errorf("ack %s: %w", id, err)
  }
  return nil
}

// Reclaim takes over messages that have waited unacknowledged for at least minIdle, for
// example because their consumer crashed. A message already delivered MaxDeliveries times is
// a poison message: it moves to the dead-letter stream instead and is acknowledged here.
func (s *Stream) Reclaim(ctx context.Context, consumer string, minIdle time.Duration, n int64) (claimed, dead []Msg, err error) {
  stale, err := s.RDB.XPendingExt(ctx, &redis.XPendingExtArgs{
    Stream: s.Key, Group: s.Group, Idle: minIdle, Start: "-", End: "+", Count: n,
  }).Result()
  if err != nil {
    return nil, nil, fmt.Errorf("list idle pending: %w", err)
  }
  counts := map[string]int64{}
  for _, p := range stale {
    if p.RetryCount < s.MaxDeliveries {
      counts[p.ID] = p.RetryCount
      continue
    }
    m, err := s.deadLetter(ctx, p)
    if err != nil {
      return nil, nil, err
    }
    dead = append(dead, m)
  }
  if len(counts) == 0 {
    return nil, dead, nil
  }
  msgs, _, err := s.RDB.XAutoClaim(ctx, &redis.XAutoClaimArgs{
    Stream: s.Key, Group: s.Group, Consumer: consumer, MinIdle: minIdle, Start: "0-0", Count: n,
  }).Result()
  if err != nil {
    return nil, nil, fmt.Errorf("claim idle pending: %w", err)
  }
  for _, m := range msgs {
    before, ok := counts[m.ID]
    if !ok {
      return nil, nil, fmt.Errorf("claimed %s, which was not idle a moment ago", m.ID)
    }
    claimed = append(claimed, Msg{ID: m.ID, Job: fmt.Sprint(m.Values["job"]), Deliveries: before + 1})
  }
  return claimed, dead, nil
}

// deadLetter copies a poison message to the dead-letter stream and acknowledges it, in one
// MULTI/EXEC block, so the message is never in both places or in neither.
func (s *Stream) deadLetter(ctx context.Context, p redis.XPendingExt) (Msg, error) {
  body, err := s.RDB.XRange(ctx, s.Key, p.ID, p.ID).Result()
  if err != nil || len(body) != 1 {
    return Msg{}, fmt.Errorf("read poison message %s: %w (found %d)", p.ID, err, len(body))
  }
  job := fmt.Sprint(body[0].Values["job"])
  if _, err := s.RDB.TxPipelined(ctx, func(tx redis.Pipeliner) error {
    tx.XAdd(ctx, &redis.XAddArgs{Stream: s.DLQ, Values: []any{"job", job, "source_id", p.ID, "deliveries", p.RetryCount}})
    tx.XAck(ctx, s.Key, s.Group, p.ID)
    return nil
  }); err != nil {
    return Msg{}, fmt.Errorf("move %s to %s: %w", p.ID, s.DLQ, err)
  }
  return Msg{ID: p.ID, Job: job, Deliveries: p.RetryCount}, nil
}

-- KEYS[1]: set of job ids already applied. KEYS[2]: hash of receipts sent per job.
-- ARGV[1]: the job id. Redis runs the whole script as one step.
if redis.call('SADD', KEYS[1], ARGV[1]) == 0 then
  return 0 -- seen before: a redelivery. Skip the effect.
end
redis.call('HINCRBY', KEYS[2], ARGV[1], 1)
return 1