Artifact [319bf172bc]

Artifact 319bf172bcf6aa6d4c1917b07086edad41cc7faed1625aa7bcc5c238c63b9366:

• File internal/chatarchive/archive_test.go — part of check-in [bd91af3158] at 2026-08-16 05:16:45 on branch trunk — chat archival: fold chat into forum-thread artifacts (0001 §5 durability)

  internal/chatarchive: batches a channel's messages (per N or T) and mints them as a 'Chat archive' forum thread through zit forum post/reply (zit#76) — one thread per channel, each batch a reply carrying the transcript, so history clones with the repo. Minting runs only on the lease holder; non-holders drop their buffers (the holder's archive folds back to every node). zitcli gains ForumPost/ForumReply/ TechnoteAdd. --chat-archive on by default. Verified live: chat batches materialize as forum threads and reply to the same thread. (user: dmestas, size: 4534) [annotate] [blame] [check-ins using]

package chatarchive

import (
	"context"
	"strings"
	"sync"
	"testing"
	"time"

	"craftdesign.group/quarry/internal/chat"
)

type fakePoster struct {
	mu      sync.Mutex
	posts   [][]byte // new-thread bodies
	replies [][]byte // reply bodies (root prefix stripped for brevity)
	nextID  int
}

func (f *fakePoster) ForumPost(title, user string, body []byte) (string, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.posts = append(f.posts, body)
	f.nextID++
	return "root", nil
}
func (f *fakePoster) ForumReply(root, user string, body []byte) (string, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.replies = append(f.replies, body)
	return "reply", nil
}
func (f *fakePoster) snap() (int, int) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.posts), len(f.replies)
}

type memRoots struct {
	mu sync.Mutex
	m  map[string]string
}

func newRoots() *memRoots { return &memRoots{m: map[string]string{}} }
func (r *memRoots) Get(code string) (string, bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	v, ok := r.m[code]
	return v, ok
}
func (r *memRoots) Set(code, root string) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.m[code] = root
	return nil
}

func holderGuard(mint func() error) (bool, error) { return true, mint() }

// A count-triggered batch mints a new thread on first flush and replies
// thereafter, always to the same channel thread.
func TestBatchByCountThenReply(t *testing.T) {
	ch := make(chan chat.Message)
	post := &fakePoster{}
	roots := newRoots()
	a := New("c1", func() (<-chan chat.Message, func()) { return ch, func() {} },
		post, roots, holderGuard, time.Hour, 2)

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { a.Run(ctx); close(done) }()

	// Two messages -> first flush -> new thread.
	ch <- chat.Message{User: "dan", Text: "hello", At: "t1"}
	ch <- chat.Message{User: "eve", Text: "hi", At: "t2"}
	// Two more -> second flush -> reply to the same thread.
	ch <- chat.Message{User: "dan", Text: "again", At: "t3"}
	ch <- chat.Message{User: "eve", Text: "yes", At: "t4"}

	waitFor(t, func() bool { p, r := post.snap(); return p == 1 && r == 1 })

	if root, ok := roots.Get("c1"); !ok || root != "root" {
		t.Fatalf("archive thread root = %q, %v", root, ok)
	}
	// The first post carries both speakers of its window.
	if !strings.Contains(string(post.posts[0]), "**dan**") || !strings.Contains(string(post.posts[0]), "**eve**") {
		t.Fatalf("transcript missing speakers:\n%s", post.posts[0])
	}
	cancel()
	<-done
}

// A timer flush archives a partial batch.
func TestBatchByTimer(t *testing.T) {
	ch := make(chan chat.Message)
	post := &fakePoster{}
	a := New("c1", func() (<-chan chat.Message, func()) { return ch, func() {} },
		post, newRoots(), holderGuard, 100*time.Millisecond, 1000)
	ctx, cancel := context.WithCancel(context.Background())
	go a.Run(ctx)
	defer cancel()

	ch <- chat.Message{User: "dan", Text: "solo", At: "t1"}
	waitFor(t, func() bool { p, _ := post.snap(); return p == 1 })
}

// On a non-holder node the buffer is dropped, not minted: the lease
// holder's archive covers those messages (no duplicate thread).
func TestNonHolderDropsBuffer(t *testing.T) {
	ch := make(chan chat.Message)
	post := &fakePoster{}
	nonHolder := func(mint func() error) (bool, error) { return false, nil }
	a := New("c1", func() (<-chan chat.Message, func()) { return ch, func() {} },
		post, newRoots(), nonHolder, time.Hour, 1)
	ctx, cancel := context.WithCancel(context.Background())
	go a.Run(ctx)
	defer cancel()

	ch <- chat.Message{User: "dan", Text: "x", At: "t1"}
	ch <- chat.Message{User: "eve", Text: "y", At: "t2"}
	time.Sleep(150 * time.Millisecond)
	if p, r := post.snap(); p != 0 || r != 0 {
		t.Fatalf("non-holder minted (posts=%d replies=%d)", p, r)
	}
}

// Remaining messages flush when Run's context is cancelled.
func TestFlushOnShutdown(t *testing.T) {
	ch := make(chan chat.Message)
	post := &fakePoster{}
	a := New("c1", func() (<-chan chat.Message, func()) { return ch, func() {} },
		post, newRoots(), holderGuard, time.Hour, 1000)
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { a.Run(ctx); close(done) }()
	ch <- chat.Message{User: "dan", Text: "buffered", At: "t1"}
	cancel()
	<-done
	if p, _ := post.snap(); p != 1 {
		t.Fatalf("shutdown did not flush buffer (posts=%d)", p)
	}
}

func waitFor(t *testing.T, cond func() bool) {
	t.Helper()
	for i := 0; i < 100; i++ {
		if cond() {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatal("condition not met in time")
}