Artifact [0c15ae46c4]

Artifact 0c15ae46c4f70da50d8f3f1d9f7d8989810da1ac4402a4f71ce821c3f392dc9a:

• File internal/chat/chat.go — part of check-in [5670d3643d] at 2026-08-16 04:50:23 on branch trunk — chat: per-repo live channel over NATS (0001 §5)

  internal/chat: messages fan out over NATS with a 7-day JetStream window (replay on join), keyed by project code so a channel is shared across nodes. chat.history/chat.post ops (post is not lease-gated — it mutates no store); web page with on-load history, a plain post form (no-JS baseline), and an SSE append stream (Datastar enhancement). Verified live: form post + cross-connection SSE delivery in-browser, and cross-node fan-out on a shared external NATS. Durable forum-thread archival waits on zit forum verbs (zit#76); the window is interim durability. (user: dmestas, size: 3899) [annotate] [blame] [check-ins using]

// Package chat is the 0001 §5 per-repository live channel: messages fan
// out over NATS to subscribers on any node, with a JetStream recent
// window (replay on join). Durability-in-the-DAG — folding batches into
// forum-thread artifacts so history clones with the repository — is the
// archiver's job and waits on zit's forum authoring verbs (zit#76); the
// window is the interim durability. A chat post mutates no store, so it
// needs no write lease: any node accepts posts and publishes them to the
// shared channel.
package chat

import (
	"encoding/json"
	"fmt"
	"time"

	"github.com/nats-io/nats.go"
)

const (
	// StreamName holds every repo's channel; subjects are chat.<code>.
	StreamName = "QUARRYCHAT"
	// window bounds the replay-on-join history per channel.
	windowAge  = 7 * 24 * time.Hour
	windowMsgs = 10_000
)

// Message is one chat line.
type Message struct {
	Repo string `json:"repo"`
	User string `json:"user"` // "anonymous" when unauthenticated
	Text string `json:"text"`
	At   string `json:"at"` // RFC3339 UTC, stamped by the receiving node
}

// Room is the chat plane over one NATS JetStream context.
type Room struct {
	nc *nats.Conn
	js nats.JetStreamContext
}

// New ensures the chat stream exists (limits retention, per-subject
// window) and returns the room. It shares the caller's NATS connection.
func New(nc *nats.Conn, js nats.JetStreamContext) (*Room, error) {
	cfg := &nats.StreamConfig{
		Name:              StreamName,
		Subjects:          []string{"chat.>"},
		Retention:         nats.LimitsPolicy,
		Discard:           nats.DiscardOld,
		MaxAge:            windowAge,
		MaxMsgsPerSubject: windowMsgs,
	}
	if _, err := js.AddStream(cfg); err != nil && err != nats.ErrStreamNameAlreadyInUse {
		// An older config (different retention/subjects) is updated in place.
		if _, uerr := js.UpdateStream(cfg); uerr != nil {
			return nil, fmt.Errorf("chat: stream: %w", err)
		}
	}
	return &Room{nc: nc, js: js}, nil
}

func subject(code string) string { return "chat." + code }

// Post stamps the message with the current UTC time and publishes it to
// the repo's channel (JetStream-persisted for the window; core
// subscribers receive it live).
func (r *Room) Post(code string, m Message) error {
	if m.User == "" {
		m.User = "anonymous"
	}
	m.At = time.Now().UTC().Format(time.RFC3339)
	raw, err := json.Marshal(m)
	if err != nil {
		return err
	}
	_, err = r.js.Publish(subject(code), raw)
	return err
}

// History returns the recent window for one channel, oldest first, at
// most limit messages.
func (r *Room) History(code string, limit int) ([]Message, error) {
	sub, err := r.js.PullSubscribe(subject(code), "",
		nats.DeliverAll(), nats.AckNone(), nats.BindStream(StreamName))
	if err != nil {
		return nil, err
	}
	defer sub.Unsubscribe()
	var out []Message
	for {
		msgs, err := sub.Fetch(256, nats.MaxWait(200*time.Millisecond))
		if err == nats.ErrTimeout {
			break
		}
		if err != nil {
			return nil, err
		}
		for _, msg := range msgs {
			var m Message
			if json.Unmarshal(msg.Data, &m) == nil {
				out = append(out, m)
			}
		}
		if len(msgs) < 256 {
			break
		}
	}
	if limit > 0 && len(out) > limit {
		out = out[len(out)-limit:]
	}
	return out, nil
}

// Subscribe delivers live messages on one channel until cancel. The
// channel buffers a little; a slow reader drops the oldest (chat is
// live, not a log — the window covers catch-up).
func (r *Room) Subscribe(code string) (<-chan Message, func()) {
	ch := make(chan Message, 32)
	sub, err := r.nc.Subscribe(subject(code), func(msg *nats.Msg) {
		var m Message
		if json.Unmarshal(msg.Data, &m) != nil {
			return
		}
		select {
		case ch <- m:
		default: // drop under backpressure
		}
	})
	if err != nil {
		close(ch)
		return ch, func() {}
	}
	r.nc.Flush() // register the SUB before returning (external NATS races)
	return ch, func() { sub.Unsubscribe() }
}