package chat
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/events"
)
// fakeFossil is a fossil server's chat surface and nothing else:
// /chat-poll answers whatever the test queued, /chat-send records what
// arrived. It enforces the two gates a real fossil enforces — HTTP
// Basic and a same-origin Referer — so a bridge that forgets either
// fails here rather than in production.
type fakeFossil struct {
mu sync.Mutex
queue []fossilMsg // handed out by /chat-poll, oldest first
sent []string // the msg= of every /chat-send
polls []string // the name= of every /chat-poll
fail int // this many polls answer 500 before any succeed
noAuth bool // answer fossil's permissions error instead
srv *httptest.Server
pollHit chan struct{}
}
func newFakeFossil(t *testing.T) *fakeFossil {
t.Helper()
f := &fakeFossil{pollHit: make(chan struct{}, 64)}
mux := http.NewServeMux()
mux.HandleFunc("/chat-poll", func(w http.ResponseWriter, r *http.Request) {
f.mu.Lock()
f.polls = append(f.polls, r.URL.Query().Get("name"))
fail, noAuth := f.fail, f.noAuth
if fail > 0 {
f.fail--
}
f.mu.Unlock()
select {
case f.pollHit <- struct{}{}:
default:
}
if !f.authed(r) {
http.Error(w, "no", http.StatusUnauthorized)
return
}
if fail > 0 {
http.Error(w, "boom", http.StatusInternalServerError)
return
}
if noAuth {
// Fossil answers a refusal INSIDE a 200 JSON envelope.
fmt.Fprint(w, `{"msgs":[{"isError":true,"xfrom":null,`+
`"xmsg":"Missing permissions or not logged in. Try <a href='/login'>logging in</a>."}]}`)
return
}
since := r.URL.Query().Get("name")
f.mu.Lock()
var out []fossilMsg
for _, m := range f.queue {
if fmt.Sprint(m.MsgID) > since || since == "0" || since == "" {
out = append(out, m)
}
}
f.queue = nil
f.mu.Unlock()
_ = json.NewEncoder(w).Encode(fossilPoll{Msgs: out})
})
mux.HandleFunc("/chat-send", func(w http.ResponseWriter, r *http.Request) {
if !f.authed(r) {
http.Error(w, "no", http.StatusUnauthorized)
return
}
_ = r.ParseForm()
f.mu.Lock()
f.sent = append(f.sent, r.FormValue("msg"))
f.mu.Unlock()
})
f.srv = httptest.NewServer(mux)
t.Cleanup(f.srv.Close)
return f
}
// authed is both gates: the credential and the Referer that makes the
// request CSRF-safe in fossil's eyes.
func (f *fakeFossil) authed(r *http.Request) bool {
user, pass, ok := r.BasicAuth()
if !ok || user != "bridgebot" || pass != "s3cret" {
return false
}
return strings.HasPrefix(r.Header.Get("Referer"), f.srv.URL)
}
func (f *fakeFossil) push(msgs ...fossilMsg) {
f.mu.Lock()
f.queue = append(f.queue, msgs...)
f.mu.Unlock()
}
func (f *fakeFossil) sentMsgs() []string {
f.mu.Lock()
defer f.mu.Unlock()
return append([]string(nil), f.sent...)
}
// memBridgeStore is the cursor and msgid map in memory.
type memBridgeStore struct {
mu sync.Mutex
cursor int64
lastEr string
seqOf map[int64]uint64
}
func newMemStore() *memBridgeStore { return &memBridgeStore{seqOf: map[int64]uint64{}} }
func (m *memBridgeStore) SetChatBridgeCursor(_ string, msgid int64, _ time.Time) error {
m.mu.Lock()
defer m.mu.Unlock()
m.cursor, m.lastEr = msgid, ""
return nil
}
func (m *memBridgeStore) RecordChatBridgeError(_ string, err error) error {
m.mu.Lock()
defer m.mu.Unlock()
if err != nil {
m.lastEr = err.Error()
}
return nil
}
func (m *memBridgeStore) MapChatBridgeMsg(_ string, msgid int64, seq uint64) error {
m.mu.Lock()
defer m.mu.Unlock()
m.seqOf[msgid] = seq
return nil
}
func (m *memBridgeStore) ChatBridgeSeq(_ string, msgid int64) (uint64, bool) {
m.mu.Lock()
defer m.mu.Unlock()
seq, ok := m.seqOf[msgid]
return seq, ok
}
func (m *memBridgeStore) err() string {
m.mu.Lock()
defer m.mu.Unlock()
return m.lastEr
}
// bridgeFixture is a real room over embedded NATS plus a fake fossil.
func bridgeFixture(t *testing.T) (*Room, *fakeFossil, *memBridgeStore, *Bridge) {
t.Helper()
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
r := room(t, ev)
f := newFakeFossil(t)
store := newMemStore()
b := NewBridge(r, store, BridgeConfig{
Repo: "alpha", Code: "codeA", URL: f.srv.URL, User: "bridgebot", Secret: "s3cret",
})
// The poll must not hang out its production timeout in a test.
b.Client = &http.Client{Timeout: 5 * time.Second}
b.SendClient = &http.Client{Timeout: 5 * time.Second}
return r, f, store, b
}
// waitFor polls cond until it holds or the test's patience runs out.
func waitFor(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for time.Now().Before(deadline) {
if cond() {
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("timed out waiting for %s", what)
}
func folded(t *testing.T, r *Room, code string) []Message {
t.Helper()
msgs, _, err := r.History(code, 0, 100)
if err != nil {
t.Fatal(err)
}
return Fold(msgs)
}
// An upstream message becomes a channel line attributed to
// `fossil:<xfrom>` with Origin "fossil" (Decision G), and the cursor
// moves to the msgid it consumed.
func TestBridgeInboundMapsAFossilMessage(t *testing.T) {
r, f, store, b := bridgeFixture(t)
f.push(fossilMsg{MsgID: 7, XFrom: "drh", XMsg: "the lease is per store"})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the message to arrive", func() bool { return len(folded(t, r, "codeA")) == 1 })
got := folded(t, r, "codeA")[0]
if got.User != "fossil:drh" || got.Origin != "fossil" || got.Kind != KindMsg {
t.Fatalf("inbound line = %+v", got)
}
if got.Text != "the lease is per store" {
t.Fatalf("inbound text = %q", got.Text)
}
waitFor(t, "the cursor to advance", func() bool {
store.mu.Lock()
defer store.mu.Unlock()
return store.cursor == 7
})
// The poll resumes from the cursor rather than re-reading.
waitFor(t, "a resumed poll", func() bool {
f.mu.Lock()
defer f.mu.Unlock()
for _, p := range f.polls {
if p == "7" {
return true
}
}
return false
})
}
// An mdel entry tombstones the message it names — found through the
// msgid map, because fossil's numbering is not quarry's.
func TestBridgeInboundDeleteBecomesATombstone(t *testing.T) {
r, f, _, b := bridgeFixture(t)
f.push(fossilMsg{MsgID: 1, XFrom: "drh", XMsg: "oops"})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the message", func() bool { return len(folded(t, r, "codeA")) == 1 })
f.push(fossilMsg{MsgID: 2, XFrom: "drh", MDel: 1})
waitFor(t, "the tombstone", func() bool {
got := folded(t, r, "codeA")
return len(got) == 1 && got[0].Deleted
})
if got := folded(t, r, "codeA")[0]; got.Text != "" {
t.Fatalf("a tombstoned message kept its text: %+v", got)
}
}
// An attachment is LINKED, never copied: fossil keeps the file, and a
// copy in a seven-day window would be a second, shorter-lived truth.
func TestBridgeInboundLinksAnAttachment(t *testing.T) {
r, f, _, b := bridgeFixture(t)
f.push(fossilMsg{MsgID: 3, XFrom: "drh", XMsg: "here it is",
FSize: 1024, FName: "patch.diff"})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the message", func() bool { return len(folded(t, r, "codeA")) == 1 })
text := folded(t, r, "codeA")[0].Text
if !strings.Contains(text, "[patch.diff]("+f.srv.URL+"/chat-download?name=3)") {
t.Fatalf("no attachment link in %q", text)
}
if !strings.Contains(text, "here it is") {
t.Fatalf("the attachment link replaced the message: %q", text)
}
}
// The loop guard: fossil attributes outbound posts to the bridge's own
// login, so its own messages must never come back in.
func TestBridgeInboundIgnoresItsOwnLogin(t *testing.T) {
r, f, _, b := bridgeFixture(t)
f.push(
fossilMsg{MsgID: 4, XFrom: "bridgebot", XMsg: "**dan:** echo of a quarry line"},
fossilMsg{MsgID: 5, XFrom: "drh", XMsg: "a real one"},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the real message", func() bool { return len(folded(t, r, "codeA")) == 1 })
// Give the skipped one every chance to appear before concluding.
time.Sleep(300 * time.Millisecond)
got := folded(t, r, "codeA")
if len(got) != 1 || got[0].Text != "a real one" {
t.Fatalf("the bridge re-imported its own message: %+v", got)
}
}
// With quarry's robot on, the upstream's timeline announcements are
// skipped — otherwise every check-in is announced twice. With it off,
// they are ordinary messages.
func TestBridgeSkipsTheUpstreamTimelineRobotOnlyWhenOursIsOn(t *testing.T) {
announcement := "Add the lease. (check-in: <a href='/info/abc'>abc</a>, user: drh)"
for _, tc := range []struct {
name string
on bool
lines int
}{
{"robot on", true, 1}, {"robot off", false, 2},
} {
t.Run(tc.name, func(t *testing.T) {
r, f, _, b := bridgeFixture(t)
b.RobotOn = func(string) bool { return tc.on }
f.push(
fossilMsg{MsgID: 8, XFrom: "timeline", XMsg: announcement},
fossilMsg{MsgID: 9, XFrom: "drh", XMsg: "a person talking"},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the person's message", func() bool {
for _, m := range folded(t, r, "codeA") {
if m.Text == "a person talking" {
return true
}
}
return false
})
time.Sleep(300 * time.Millisecond)
if got := folded(t, r, "codeA"); len(got) != tc.lines {
t.Fatalf("channel holds %d lines, want %d: %+v", len(got), tc.lines, got)
}
})
}
}
// Outbound: a person's message is posted upstream with the name
// prefix fossil has no other place for. A robot line and a line that
// came FROM fossil are never forwarded.
func TestBridgeOutboundPrefixesAndRefusesEchoes(t *testing.T) {
r, f, _, b := bridgeFixture(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
// Let the tail attach before anything is posted: it starts at NOW,
// so a message published first is correctly not forwarded.
time.Sleep(300 * time.Millisecond)
for _, m := range []Message{
{Repo: "alpha", User: "dan", Text: "hello fossil"},
{Repo: "alpha", User: "robot", Kind: KindRobot, Origin: "robot", Text: "check-in abc"},
{Repo: "alpha", User: "fossil:drh", Origin: "fossil", Text: "came from over there"},
} {
if _, err := r.Post("codeA", m); err != nil {
t.Fatal(err)
}
}
waitFor(t, "the outbound post", func() bool { return len(f.sentMsgs()) > 0 })
time.Sleep(300 * time.Millisecond)
sent := f.sentMsgs()
if len(sent) != 1 {
t.Fatalf("the bridge forwarded %d messages, want 1: %q", len(sent), sent)
}
if sent[0] != "**dan:** hello fossil" {
t.Fatalf("outbound msg = %q", sent[0])
}
}
// A failing upstream is recorded and retried, not spun on: the first
// poll after a 5xx does not happen immediately.
func TestBridgeBacksOffOnFailure(t *testing.T) {
_, f, store, b := bridgeFixture(t)
f.mu.Lock()
f.fail = 3
f.mu.Unlock()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the failure to be recorded", func() bool { return store.err() != "" })
if !strings.Contains(store.err(), "500") {
t.Fatalf("recorded error does not name the status: %q", store.err())
}
// backoffMin is 30 s, so within a second of the first failure there
// must not have been a second poll.
f.mu.Lock()
n := len(f.polls)
f.mu.Unlock()
time.Sleep(time.Second)
f.mu.Lock()
after := len(f.polls)
f.mu.Unlock()
if after > n {
t.Fatalf("the bridge retried %d times inside a second: no backoff", after-n)
}
}
// A refusal fossil delivers inside a 200 JSON envelope is an error,
// not a message: importing it would put "Missing permissions" in the
// channel as if somebody had said it.
func TestBridgeTreatsAnEnvelopedRefusalAsAnError(t *testing.T) {
r, f, store, b := bridgeFixture(t)
f.mu.Lock()
f.noAuth = true
f.mu.Unlock()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the refusal", func() bool { return store.err() != "" })
if !strings.Contains(store.err(), "Missing permissions") {
t.Fatalf("refusal not reported in words: %q", store.err())
}
if strings.Contains(store.err(), "<a href") {
t.Fatalf("the refusal was logged with fossil's markup in it: %q", store.err())
}
if got := folded(t, r, "codeA"); len(got) != 0 {
t.Fatalf("a refusal was imported as a message: %+v", got)
}
}
// A long poll that expires with nothing new answers an empty body.
// That is the quiet case, not a broken upstream: reading it as an
// error would put every idle channel into permanent backoff.
func TestBridgeTreatsAnEmptyBodyAsQuiet(t *testing.T) {
_, f, store, b := bridgeFixture(t)
quiet := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/chat-poll" {
select {
case f.pollHit <- struct{}{}:
default:
}
}
// Nothing at all, exactly as fossil answers an expired poll.
}))
defer quiet.Close()
b.cfg.URL = quiet.URL
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
// Several polls happen inside a couple of seconds, which they
// could not if the first had been recorded as a failure.
for i := 0; i < 2; i++ {
select {
case <-f.pollHit:
case <-time.After(5 * time.Second):
t.Fatal("the bridge stopped polling a quiet upstream")
}
}
if e := store.err(); e != "" {
t.Fatalf("a quiet poll was recorded as a failure: %q", e)
}
}