News feed

// Post stores the post, adds it to the author's own list, and pushes it to the timelines of
// active followers, unless the author has Threshold followers or more. It returns the post ID and
// the number of timelines written.
func (s *Service) Post(ctx context.Context, author int64, body string) (int64, int, error) {
  id, err := s.IDs.Next()
  if err != nil {
    return 0, 0, fmt.Errorf("post id: %w", err)
  }
  var followers int64
  if err := s.DB.QueryRow(ctx, Q("insert_post"), id, author, body).Scan(&followers); err != nil {
    return 0, 0, fmt.Errorf("insert post: %w", err)
  }
  n, err := s.Deliver(ctx, author, id, followers)
  return id, n, err
}

// Deliver puts a stored post into the author's own list, then into followers' timelines unless
// the author is pulled.
func (s *Service) Deliver(ctx context.Context, author, id, followers int64) (int, error) {
  if err := s.add(ctx, s.authored(author), id); err != nil {
    return 0, err
  }
  if followers >= s.Threshold {
    return 0, nil // a large account: readers pull its posts
  }
  return s.FanOut(ctx, author, id)
}

// FanOut writes one entry into the timeline of every active follower, in pipelined batches. Each
// timeline is trimmed to Cap entries in the same round trip.
func (s *Service) FanOut(ctx context.Context, author, id int64) (int, error) {
  targets, err := s.followers(ctx, author, s.ActiveSince)
  if err != nil {
    return 0, err
  }
  for chunk := range slices.Chunk(targets, max(s.Batch, 1)) {
    _, err := s.RDB.Pipelined(ctx, func(p redis.Pipeliner) error {
      for _, f := range chunk {
        key := s.timeline(f)
        p.ZAdd(ctx, key, redis.Z{Score: score(id), Member: member(id)})
        p.ZRemRangeByRank(ctx, key, 0, int64(-s.Cap-1))
      }
      return nil
    })
    if err != nil {
      return 0, fmt.Errorf("fan out post %d: %w", id, err)
    }
  }
  return len(targets), nil
}

// Page returns up to n post IDs older than the cursor, newest first. The cursor is the last ID of
// the previous page (0 for the first page). Sources are the reader's timeline plus the own list of
// every pulled account the reader follows; one pipelined round trip reads them all.
func (s *Service) Page(ctx context.Context, reader, cursor int64, n int) ([]Entry, error) {
  pulled, err := s.pulled(ctx, reader)
  if err != nil {
    return nil, err
  }
  keys := []string{s.timeline(reader)}
  for _, a := range pulled {
    keys = append(keys, s.authored(a))
  }
  upTo := upTo(cursor)
  cmds := make([]*redis.StringSliceCmd, len(keys))
  _, err = s.RDB.Pipelined(ctx, func(p redis.Pipeliner) error {
    for i, k := range keys {
      cmds[i] = p.ZRangeArgs(ctx, newest(k, upTo, 0, n))
    }
    return nil
  })
  if err != nil {
    return nil, fmt.Errorf("read feed of %d: %w", reader, err)
  }
  var merged []Entry
  for i, c := range cmds {
    ids, err := below(c.Val(), cursor, n)
    if err != nil {
      return nil, err
    }
    if len(ids) < n && len(c.Val()) == n { // IDs in the cursor's millisecond used up the read
      if ids, err = s.older(ctx, keys[i], cursor, n); err != nil {
        return nil, err
      }
    }
    merged = append(merged, ids...)
  }
  slices.SortFunc(merged, func(a, b Entry) int { return cmp.Compare(b, a) })
  merged = slices.Compact(merged)
  return merged[:min(n, len(merged))], nil
}

// newest is ZRANGE key max -inf BYSCORE REV LIMIT offset n: the newest entries at or below max.
// Before Redis 6.2 the same command was ZREVRANGEBYSCORE.
func newest(key, maxScore string, offset int64, n int) redis.ZRangeArgs {
  return redis.ZRangeArgs{Key: key, Start: maxScore, Stop: "-inf", ByScore: true, Rev: true, Offset: offset, Count: int64(n)}
}

// upTo is the highest score a page may read: the cursor's millisecond, inclusive, because other
// posts of that millisecond may sort below the cursor.
func upTo(cursor int64) string {
  if cursor == 0 {
    return "+inf"
  }
  return strconv.FormatFloat(score(cursor), 'f', 0, 64)
}

// below keeps the members whose ID is lower than the cursor, up to n.
func below(members []string, cursor int64, n int) ([]Entry, error) {
  var out []Entry
  for _, m := range members {
    id, err := strconv.ParseInt(m, 10, 64)
    if err != nil {
      return nil, fmt.Errorf("timeline member %q: %w", m, err)
    }
    if (cursor == 0 || id < cursor) && len(out) < n {
      out = append(out, id)
    }
  }
  return out, nil
}

// older reads one sorted set page by page from the cursor's millisecond until it has n IDs below
// the cursor. Page needs it only when one millisecond holds n posts or more.
func (s *Service) older(ctx context.Context, key string, cursor int64, n int) ([]Entry, error) {
  var out []Entry
  for offset := int64(0); len(out) < n; {
    got, err := s.RDB.ZRangeArgs(ctx, newest(key, upTo(cursor), offset, n)).Result()
    if err != nil {
      return nil, fmt.Errorf("read %s: %w", key, err)
    }
    ids, err := below(got, cursor, n-len(out))
    if err != nil {
      return nil, err
    }
    out = append(out, ids...)
    if len(got) < n {
      break
    }
    offset += int64(len(got))
  }
  return out, nil
}