package chat
import (
"encoding/json"
"log"
"sort"
"sync"
"time"
"github.com/nats-io/nats.go"
)
// Presence is who is reading a channel right now.
//
// "Here" means one thing: an open live stream on this channel. Every
// signed-in reader holds one, the server knows exactly when it opens and
// closes, and that is the whole definition — no idle states, no "away
// after five minutes", because those are the first thing every chat
// client adds and the first thing nobody trusts. A dot or no dot.
//
// It rides CORE NATS, not JetStream, on purpose. Presence is the one
// thing that must not be durable: a heartbeat from an hour ago means
// nothing, and a replayed one would put a person in a room they left.
// Every quarry instance subscribes to the same subjects, so two servers
// behind a balancer agree on who is here without talking to each other.
type Presence struct {
nc *nats.Conn
sub *nats.Subscription
mu sync.Mutex
// seen is code -> user -> last heartbeat. A user with no fresh
// heartbeat is gone whether or not their leave ever arrived — a
// crashed server sends no goodbye.
seen map[string]map[string]time.Time
// watchers are told, non-blockingly, when a channel's set changes.
watchers map[string]map[chan struct{}]struct{}
done chan struct{}
}
const (
// presenceBeat is how often an open stream says "still here".
presenceBeat = 20 * time.Second
// presenceStale is how long silence means "gone": two missed beats
// plus slack, so one dropped packet does not flicker a person out.
presenceStale = 50 * time.Second
presenceSweep = 10 * time.Second
)
// presenceMsg is one heartbeat or goodbye on the wire.
type presenceMsg struct {
User string `json:"u"`
Bye bool `json:"bye,omitempty"`
}
func presenceSubject(code string) string { return "presence." + code }
// NewPresence subscribes to every channel's presence and starts the
// sweep that turns silence into absence.
func NewPresence(nc *nats.Conn) (*Presence, error) {
p := &Presence{
nc: nc,
seen: map[string]map[string]time.Time{},
watchers: map[string]map[chan struct{}]struct{}{},
done: make(chan struct{}),
}
sub, err := nc.Subscribe("presence.>", p.receive)
if err != nil {
return nil, err
}
p.sub = sub
go p.sweep()
return p, nil
}
// Close stops the sweep and the subscription.
func (p *Presence) Close() {
close(p.done)
if p.sub != nil {
_ = p.sub.Unsubscribe()
}
}
func (p *Presence) receive(m *nats.Msg) {
code := m.Subject[len("presence."):]
var pm presenceMsg
if json.Unmarshal(m.Data, &pm) != nil || pm.User == "" {
return
}
p.mu.Lock()
users := p.seen[code]
if users == nil {
users = map[string]time.Time{}
p.seen[code] = users
}
_, was := users[pm.User]
if pm.Bye {
delete(users, pm.User)
} else {
users[pm.User] = time.Now()
}
changed := was == pm.Bye // arrived when absent, or left when present
p.mu.Unlock()
if changed {
p.notify(code)
}
}
// sweep is what makes a crash a departure: nobody said goodbye, but
// nobody has said hello for presenceStale either.
func (p *Presence) sweep() {
t := time.NewTicker(presenceSweep)
defer t.Stop()
for {
select {
case <-p.done:
return
case now := <-t.C:
var changed []string
p.mu.Lock()
for code, users := range p.seen {
for u, at := range users {
if now.Sub(at) > presenceStale {
delete(users, u)
changed = append(changed, code)
}
}
}
p.mu.Unlock()
for _, code := range changed {
p.notify(code)
}
}
}
}
func (p *Presence) notify(code string) {
p.mu.Lock()
ws := p.watchers[code]
p.mu.Unlock()
for ch := range ws {
select {
case ch <- struct{}{}:
default: // a watcher that has not drained its last tick needs no second one
}
}
}
// Join announces user on code, keeps announcing until leave is called,
// and says goodbye then. The first announcement is immediate, so the
// room shows the arrival before the first beat.
func (p *Presence) Join(code, user string) (leave func()) {
if user == "" {
return func() {}
}
say := func(bye bool) {
b, _ := json.Marshal(presenceMsg{User: user, Bye: bye})
if err := p.nc.Publish(presenceSubject(code), b); err != nil {
log.Printf("chat: presence %s %s: %v", code, user, err)
}
}
say(false)
stop := make(chan struct{})
var once sync.Once
go func() {
t := time.NewTicker(presenceBeat)
defer t.Stop()
for {
select {
case <-stop:
return
case <-t.C:
say(false)
}
}
}()
return func() {
once.Do(func() {
close(stop)
say(true)
})
}
}
// Present is who is here, sorted, so two renders of the same room draw
// the same list.
func (p *Presence) Present(code string) []string {
p.mu.Lock()
defer p.mu.Unlock()
users := p.seen[code]
out := make([]string, 0, len(users))
now := time.Now()
for u, at := range users {
if now.Sub(at) <= presenceStale {
out = append(out, u)
}
}
sort.Strings(out)
return out
}
// Changed ticks whenever code's set changes: an arrival, a departure,
// or silence becoming absence. Buffered by one, so a slow consumer
// sees "something changed" rather than a queue of every change.
func (p *Presence) Changed(code string) (<-chan struct{}, func()) {
ch := make(chan struct{}, 1)
p.mu.Lock()
ws := p.watchers[code]
if ws == nil {
ws = map[chan struct{}]struct{}{}
p.watchers[code] = ws
}
ws[ch] = struct{}{}
p.mu.Unlock()
return ch, func() {
p.mu.Lock()
delete(p.watchers[code], ch)
p.mu.Unlock()
}
}