// bridge.go is RFC 0011 §9: an opt-in connector between one quarry
// channel and one Fossil server's chat.
//
// Fossil's chat is a server-local table, never synced, gated by `C`,
// long-polled through /chat-poll and written through /chat-send. Every
// key below was checked against a real fossil 2.28 (`msgid`, `mtime`,
// `xfrom`, `xmsg`, `fsize`, `fname`, `mdel`) rather than assumed.
//
// Two constraints that a reading of chat.c alone does not surface, and
// that a bridge silently fails without:
//
// 1. HTTP Basic authentication is refused unless the upstream has the
// `http_authentication_ok` config entry set. Without it every
// request is anonymous and every answer is a permissions error.
// 2. /chat-poll creates its full-text TEMP triggers on the way in, so
// fossil's CSRF gate must consider the request safe or the poll
// dies with "not authorized: CREATE TEMP TRIGGER". A same-origin
// `Referer` header is what makes it safe, so every request here
// carries one.
//
// What the bridge does not pretend: fossil messages cannot be edited
// from quarry, two fossil servers are two chats, and a bridged
// channel's history is still quarry's seven-day window.
package chat
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net/http"
"net/url"
"regexp"
"strings"
"time"
)
const (
// pollTimeout must exceed fossil's chat-poll-timeout (420 s by
// default) or every poll is a client-side cancellation that reads
// nothing and looks, from the log, like an upstream that is down.
pollTimeout = 9 * time.Minute
// sendTimeout is an ordinary write; nothing about it hangs.
sendTimeout = 30 * time.Second
// backoffMin/backoffMax pace retries after a failed poll. Doubling
// from half a minute to five keeps a broken upstream from being
// hammered while a transient one recovers within a minute.
backoffMin = 30 * time.Second
backoffMax = 5 * time.Minute
)
// BridgeConfig is one bridged channel: which repository, which channel
// key, and the upstream credential the poller replays.
type BridgeConfig struct {
Repo string
Code string
URL string
User string
Secret string
// LastMsgID is where the poller resumes; 0 asks fossil for the
// channel's recent history rather than for nothing.
LastMsgID int64
}
// BridgeStore is the cursor and the msgid map. Both are OLTP, both are
// window-bounded, and a nil store is a bridge with no memory: it still
// carries messages, it just re-reads the upstream from where it
// started every time the process restarts.
type BridgeStore interface {
// SetChatBridgeCursor advances how far the poller has read.
SetChatBridgeCursor(repo string, msgid int64, at time.Time) error
// RecordChatBridgeError remembers why the last poll failed.
RecordChatBridgeError(repo string, err error) error
// MapChatBridgeMsg remembers which sequence an upstream msgid
// became, so a later mdel can tombstone the right message.
MapChatBridgeMsg(repo string, msgid int64, seq uint64) error
// ChatBridgeSeq answers that question back.
ChatBridgeSeq(repo string, msgid int64) (uint64, bool)
}
// Bridge carries one channel both ways.
type Bridge struct {
room *Room
store BridgeStore
cfg BridgeConfig
// Client is the HTTP client the poller uses. Its timeout must
// exceed the upstream's poll hang; NewBridge sets one that does.
Client *http.Client
// SendClient is the short-timeout client for outbound posts.
SendClient *http.Client
// RobotOn reports whether this repository's own activity robot is
// on. When it is, the upstream's timeline announcements are
// skipped: otherwise every check-in is announced twice, once by
// each robot, which is the failure that teaches people to mute the
// channel. nil means "off", the direction this must fail in.
RobotOn func(repo string) bool
// timelineUser is the upstream's chat-timeline-user when it could
// be learned; "" falls back to the shape heuristic below.
timelineUser string
}
func NewBridge(room *Room, store BridgeStore, cfg BridgeConfig) *Bridge {
return &Bridge{
room: room, store: store, cfg: cfg,
Client: &http.Client{Timeout: pollTimeout},
SendClient: &http.Client{Timeout: sendTimeout},
}
}
// fossilMsg is one /chat-poll entry, field for field as fossil emits
// it. Anything not named here is fossil's presentation (uclr, lmtime)
// and has no meaning on this side.
type fossilMsg struct {
MsgID int64 `json:"msgid"`
MTime string `json:"mtime"`
XFrom string `json:"xfrom"`
XMsg string `json:"xmsg"`
FSize int64 `json:"fsize"`
FName string `json:"fname"`
MDel int64 `json:"mdel"`
IsError bool `json:"isError"`
}
type fossilPoll struct {
Msgs []fossilMsg `json:"msgs"`
}
// Run carries the channel both ways until ctx is done. Inbound is a
// long poll; outbound is a tail of the channel from NOW — a bridge
// starting up must not replay the window into the upstream, which
// would post a week of quarry's conversation as new fossil messages.
func (b *Bridge) Run(ctx context.Context) {
if b == nil || b.room == nil || b.cfg.URL == "" {
return
}
b.timelineUser = b.fetchTimelineUser(ctx)
go b.outbound(ctx)
b.inbound(ctx)
}
func (b *Bridge) inbound(ctx context.Context) {
backoff := backoffMin
for ctx.Err() == nil {
n, err := b.pollOnce(ctx)
if err != nil {
if ctx.Err() != nil {
return
}
log.Printf("chat: bridge %s: poll: %v", b.cfg.Repo, err)
if b.store != nil {
if serr := b.store.RecordChatBridgeError(b.cfg.Repo, err); serr != nil {
log.Printf("chat: bridge %s: recording the failure: %v", b.cfg.Repo, serr)
}
}
select {
case <-ctx.Done():
return
case <-time.After(backoff):
}
if backoff *= 2; backoff > backoffMax {
backoff = backoffMax
}
continue
}
backoff = backoffMin
if n == 0 {
// A poll that hung out its timeout and returned nothing is
// the ordinary quiet case, not a reason to pause — but a
// server answering instantly with nothing would spin, so
// the loop yields briefly either way.
select {
case <-ctx.Done():
return
case <-time.After(time.Second):
}
}
}
}
// pollOnce reads one /chat-poll response into the channel and answers
// how many upstream entries it consumed.
func (b *Bridge) pollOnce(ctx context.Context) (int, error) {
poll := fmt.Sprintf("%s/chat-poll?name=%d&raw=1", strings.TrimRight(b.cfg.URL, "/"), b.cfg.LastMsgID)
body, err := b.get(ctx, poll)
if err != nil {
return 0, err
}
// A poll that hangs out its timeout with nothing new answers an
// EMPTY BODY, not an empty envelope (verified against fossil 2.28).
// Reading that as malformed JSON would turn every quiet interval
// into a recorded failure and a thirty-second backoff — a bridge
// that works only on a busy channel.
if len(strings.TrimSpace(string(body))) == 0 {
return 0, nil
}
var out fossilPoll
if err := json.Unmarshal(body, &out); err != nil {
// A fossil that refuses answers HTML, not JSON. Saying so is
// more use than "invalid character '<'".
return 0, fmt.Errorf("chat-poll did not answer JSON (upstream refused or is not a fossil server): %w", err)
}
for _, m := range out.Msgs {
if m.IsError {
return 0, errors.New("chat-poll: " + plainText(m.XMsg))
}
b.consume(m)
}
return len(out.Msgs), nil
}
// consume turns one upstream entry into a channel line, advancing the
// cursor whatever the entry turned out to be: a message the bridge
// deliberately skips has still been READ, and leaving the cursor
// behind it would re-read it on every poll for ever.
func (b *Bridge) consume(m fossilMsg) {
defer b.advance(m.MsgID)
switch {
case m.MDel != 0:
// A deletion names the msgid it removed. Without a mapping the
// message is either one the bridge never imported or one the
// window has already dropped, and in both cases there is
// nothing here to tombstone.
if b.store == nil {
return
}
seq, ok := b.store.ChatBridgeSeq(b.cfg.Repo, m.MDel)
if !ok {
return
}
if _, err := b.room.Post(b.cfg.Code, Message{
Repo: b.cfg.Repo, Kind: KindDel, Ref: seq,
Origin: "fossil", User: bridgeUser(m.XFrom),
}); err != nil {
log.Printf("chat: bridge %s: tombstone for #%d: %v", b.cfg.Repo, seq, err)
}
case b.skip(m):
return
default:
text := b.textOf(m)
if strings.TrimSpace(text) == "" {
return
}
seq, err := b.room.Post(b.cfg.Code, Message{
Repo: b.cfg.Repo, Kind: KindMsg, Origin: "fossil",
User: bridgeUser(m.XFrom), Text: text,
})
if err != nil {
log.Printf("chat: bridge %s: importing msgid %d: %v", b.cfg.Repo, m.MsgID, err)
return
}
if b.store != nil {
if err := b.store.MapChatBridgeMsg(b.cfg.Repo, m.MsgID, seq); err != nil {
log.Printf("chat: bridge %s: mapping msgid %d: %v", b.cfg.Repo, m.MsgID, err)
}
}
}
}
func (b *Bridge) advance(msgid int64) {
if msgid <= b.cfg.LastMsgID {
return
}
b.cfg.LastMsgID = msgid
if b.store != nil {
if err := b.store.SetChatBridgeCursor(b.cfg.Repo, msgid, time.Now()); err != nil {
log.Printf("chat: bridge %s: cursor: %v", b.cfg.Repo, err)
}
}
}
// skip is the two reasons an inbound message is not carried.
//
// The loop guard is first and unconditional: outbound posts arrive at
// the upstream under the bridge's own login, so re-importing them
// would echo every quarry message back into quarry.
//
// The second is the upstream's timeline robot, skipped only while
// quarry's own robot is announcing the same events (§10). Fossil does
// not expose `chat-timeline-user` on any endpoint a bridge login can
// read, so the name is used when /json/config volunteered it and the
// announcement's own SHAPE is used when it did not: fossil builds
// those messages with a trailing "(check-in: <a href=…" clause that no
// hand-typed message carries.
func (b *Bridge) skip(m fossilMsg) bool {
if m.XFrom == "" || strings.EqualFold(m.XFrom, b.cfg.User) {
return true
}
robotOn := b.RobotOn != nil && b.RobotOn(b.cfg.Repo)
if !robotOn {
return false
}
if b.timelineUser != "" {
return strings.EqualFold(m.XFrom, b.timelineUser)
}
return timelineShape.MatchString(m.XMsg)
}
// timelineShape is fossil's chat_msg_from_event() format: the event's
// own words followed by a parenthesised artifact clause it builds as
// HTML even in a raw message.
var timelineShape = regexp.MustCompile(`\((check-in|artifact|ticket|wiki|forum post|event): <a href=`)
// textOf is the message body as the channel should hold it: fossil's
// markdown verbatim, plus a link to any attachment. Fossil keeps the
// file itself; the bridge links it rather than copying it, because a
// copy in the channel window would be a second, shorter-lived truth.
func (b *Bridge) textOf(m fossilMsg) string {
text := m.XMsg
if m.FSize <= 0 {
return text
}
name := m.FName
if name == "" {
name = "attachment"
}
link := fmt.Sprintf("[%s](%s/chat-download?name=%d) (%d bytes)",
name, strings.TrimRight(b.cfg.URL, "/"), m.MsgID, m.FSize)
if strings.TrimSpace(text) == "" {
return link
}
return text + "\n\n" + link
}
// bridgeUser is Decision G: a bridged line shows as `fossil:<xfrom>`.
// There is no account mapping — the prefix is the honest statement
// that this name belongs to another server's user table, not to a
// quarry account somebody could be confused for.
func bridgeUser(xfrom string) string {
if xfrom == "" {
return "fossil:anonymous"
}
return "fossil:" + xfrom
}
// outbound posts quarry's own messages upstream. Only a person's:
// robot lines are quarry's announcements about quarry, and a
// fossil-origin line came FROM there — echoing either back is the loop
// this half has to not be.
func (b *Bridge) outbound(ctx context.Context) {
msgs, cancel := b.room.Tail(b.cfg.Code, 0)
defer cancel()
for {
select {
case <-ctx.Done():
return
case m, ok := <-msgs:
if !ok {
return
}
if m.Origin != "" || m.Kind != KindMsg || strings.TrimSpace(m.Text) == "" {
continue
}
if err := b.send(ctx, m); err != nil {
log.Printf("chat: bridge %s: sending #%d: %v", b.cfg.Repo, m.Seq, err)
}
}
}
}
// send posts one quarry message to /chat-send. Fossil attributes
// everything to the login that sent it, so the author's name is
// carried in the text — the only place the protocol leaves for it.
func (b *Bridge) send(ctx context.Context, m Message) error {
user := m.User
if user == "" {
user = "anonymous"
}
form := url.Values{"msg": {"**" + user + ":** " + m.Text}}
req, err := http.NewRequestWithContext(ctx, "POST",
strings.TrimRight(b.cfg.URL, "/")+"/chat-send", strings.NewReader(form.Encode()))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
b.arm(req)
resp, err := b.SendClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("chat-send: %s", resp.Status)
}
// A refusal is a 200 carrying a JSON error object; success is an
// empty body. Reading the status alone would call every rejected
// message delivered.
if e := errorIn(body); e != "" {
return errors.New("chat-send: " + e)
}
return nil
}
func (b *Bridge) get(ctx context.Context, u string) ([]byte, error) {
req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
if err != nil {
return nil, err
}
b.arm(req)
resp, err := b.Client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("%s: %s", u, resp.Status)
}
return body, nil
}
// arm puts the credential and the same-origin Referer on a request.
// The Referer is not decoration: fossil treats a request without one
// as CSRF-unsafe and refuses /chat-poll's own table setup, which
// surfaces as a database error rather than as an auth failure.
func (b *Bridge) arm(req *http.Request) {
if b.cfg.User != "" {
req.SetBasicAuth(b.cfg.User, b.cfg.Secret)
}
req.Header.Set("Referer", strings.TrimRight(b.cfg.URL, "/")+"/chat")
}
// fetchTimelineUser asks the upstream for its chat-timeline-user once,
// at startup. Most fossil builds ship without the JSON API and answer
// HTML or 404; that is expected, costs one request, and leaves the
// shape heuristic in charge.
func (b *Bridge) fetchTimelineUser(ctx context.Context) string {
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
body, err := b.get(ctx, strings.TrimRight(b.cfg.URL, "/")+"/json/config/chat-timeline-user")
if err != nil {
return ""
}
var out struct {
Payload map[string]any `json:"payload"`
}
if json.Unmarshal(body, &out) != nil {
return ""
}
name, _ := out.Payload["chat-timeline-user"].(string)
return name
}
// errorIn answers the message inside a fossil JSON error object, "" if
// the body is not one.
func errorIn(body []byte) string {
var e struct {
IsError bool `json:"isError"`
XMsg string `json:"xmsg"`
}
if json.Unmarshal(body, &e) != nil || !e.IsError {
return ""
}
return plainText(e.XMsg)
}
// plainText strips the anchor markup fossil puts in its own error
// strings, so a log line reads as a sentence.
func plainText(s string) string {
var b strings.Builder
depth := 0
for _, r := range s {
switch {
case r == '<':
depth++
case r == '>' && depth > 0:
depth--
case depth == 0:
b.WriteRune(r)
}
}
return strings.TrimSpace(b.String())
}