Chat and presence

// route looks up where the user's devices are connected and publishes one envelope per gateway.
// A gateway that receives nothing is dead: its entries leave the registry. A user with no live
// device gets a push notification for a message; the message itself waits in Postgres.
func (g *Gateway) route(ctx context.Context, user, except string, f Frame) error {
  devices, err := g.rdb.HGetAll(ctx, g.cfg.sessKey(user)).Result()
  if err != nil {
    return fmt.Errorf("registry %s: %w", user, err)
  }
  byGW := map[string][]string{}
  for d, gw := range devices {
    if d != except {
      byGW[gw] = append(byGW[gw], d)
    }
  }
  delivered := false
  for gw, ds := range byGW {
    slices.Sort(ds)
    p, err := json.Marshal(envelope{User: user, Devices: ds, Frame: f})
    if err != nil {
      return fmt.Errorf("marshal envelope: %w", err)
    }
    n, err := g.rdb.Publish(ctx, g.cfg.gwChan(gw), p).Result()
    if err != nil {
      return fmt.Errorf("publish to %s: %w", gw, err)
    }
    g.Stats.Published.Add(1)
    if n > 0 {
      delivered = true
      continue
    }
    g.Stats.DeadRoutes.Add(1)
    args := append([]any{gw}, anys(ds)...)
    if err := dropStale.Run(ctx, g.rdb, []string{g.cfg.sessKey(user)}, args...).Err(); err != nil {
      return fmt.Errorf("drop stale entries of %s: %w", user, err)
    }
  }
  if !delivered && f.T == "msg" && except == "" {
    if err := g.rdb.RPush(ctx, g.cfg.pushKey(user), fmt.Sprintf("%d:%d", f.Conv, f.Seq)).Err(); err != nil {
      return fmt.Errorf("queue push for %s: %w", user, err)
    }
    g.Stats.Pushes.Add(1)
  }
  return nil
}

// send stores the message with the next sequence number of its conversation, acks the sender,
// and routes the message to every device of every member. A retry with the same client message
// ID gets the first sequence number back and is not routed again.
func (g *Gateway) send(ctx context.Context, s *session, f Frame) error {
  seq, dup, err := g.store(ctx, f.Conv, s.user, f.CMID, f.Body)
  if err != nil {
    return err
  }
  g.reply(s, Frame{T: "ack", Conv: f.Conv, CMID: f.CMID, Seq: seq, Dup: dup})
  if dup {
    return nil
  }
  msg := Frame{T: "msg", Conv: f.Conv, Seq: seq, From: s.user, CMID: f.CMID, Body: f.Body}
  members, err := g.membersOf(ctx, f.Conv)
  if err != nil {
    return err
  }
  for _, m := range members {
    except := ""
    if m == s.user {
      except = s.device // the sender's other devices get a copy
    }
    if err := g.route(ctx, m, except, msg); err != nil {
      return err
    }
  }
  return nil
}

// store takes the next sequence number under the conversation's row lock. The lock also orders
// two copies of one retry, so the second finds the first and reuses its number.
func (g *Gateway) store(ctx context.Context, conv int64, sender, cmid, body string) (seq int64, dup bool, err error) {
  tx, err := g.db.Begin(ctx)
  if err != nil {
    return 0, false, fmt.Errorf("begin: %w", err)
  }
  defer func() {
    if err != nil {
      err = errors.Join(err, tx.Rollback(ctx))
    }
  }()
  if err := tx.QueryRow(ctx, Q("next_seq"), conv, sender).Scan(&seq); err != nil {
    if errors.Is(err, pgx.ErrNoRows) {
      return 0, false, fmt.Errorf("%s is not in conversation %d", sender, conv)
    }
    return 0, false, fmt.Errorf("next seq: %w", err)
  }
  var first int64
  switch err := tx.QueryRow(ctx, Q("find_sent"), conv, sender, cmid).Scan(&first); {
  case err == nil:
    return first, true, tx.Rollback(ctx) // a retry: give back the number just taken
  case !errors.Is(err, pgx.ErrNoRows):
    return 0, false, fmt.Errorf("find sent: %w", err)
  }
  if _, err := tx.Exec(ctx, Q("insert_message"), conv, seq, sender, cmid, body); err != nil {
    return 0, false, fmt.Errorf("insert message: %w", err)
  }
  if err := tx.Commit(ctx); err != nil {
    return 0, false, fmt.Errorf("commit: %w", err)
  }
  return seq, false, nil
}

// heartbeat renews the registry entry and the presence key. Each is a lease: without a
// heartbeat it expires, so a crashed gateway's users drop out by themselves. Only a change from
// offline to online is published.
func (g *Gateway) heartbeat(ctx context.Context, s *session) error {
  if err := g.rdb.Expire(ctx, g.cfg.sessKey(s.user), g.cfg.SessionTTL).Err(); err != nil {
    return fmt.Errorf("renew session %s: %w", s.key(), err)
  }
  err := g.rdb.SetArgs(ctx, g.cfg.presKey(s.user), g.ID, redis.SetArgs{TTL: g.cfg.PresenceTTL, Get: true}).Err()
  if errors.Is(err, redis.Nil) { // there was no key: the user just came online
    return g.publishPresence(ctx, s.user, "online")
  }
  if err != nil {
    return fmt.Errorf("presence %s: %w", s.user, err)
  }
  return nil
}

// watch subscribes this gateway to the presence of some users, for one session, and replies
// with their state now. Clients watch only the people on screen.
func (g *Gateway) watch(ctx context.Context, s *session, users []string) error {
  for _, u := range users {
    g.mu.Lock()
    first := len(g.watchers[u]) == 0
    if first {
      g.watchers[u] = map[*session]bool{}
    }
    g.watchers[u][s] = true
    g.mu.Unlock()
    if first {
      if err := g.sub.Subscribe(ctx, g.cfg.presChan(u)); err != nil {
        return fmt.Errorf("watch %s: %w", u, err)
      }
    }
    n, err := g.rdb.Exists(ctx, g.cfg.presKey(u)).Result()
    if err != nil {
      return fmt.Errorf("presence of %s: %w", u, err)
    }
    g.reply(s, Frame{T: "presence", User: u, State: map[bool]string{true: "online", false: "offline"}[n == 1]})
  }
  return nil
}