// chatbridge.go stores RFC 0011 §9's opt-in Fossil chat bridge: which
// upstream a repository's channel is bridged to, the credential the
// poller signs in with, and how far it has read.
//
// It sits beside `mirror` for the same reason the mirror row does —
// this is forge configuration, not DAG state, and it must never sync.
// A bridged channel is still quarry's window; the bridge is a
// connector, and losing this table loses the connection, not history.
//
// The credential is stored the way sync_user secrets are: a plain
// column, because the poller has to REPLAY it at every request (HTTP
// Basic against someone else's fossil), and a one-way hash cannot be
// replayed. Nothing reads it back out to a caller — the ops surface
// has no field for it, and the only reader is the bridge goroutine.
package oltp
import (
"database/sql"
"errors"
"fmt"
"net/url"
"regexp"
"time"
)
const chatBridgeSchema = `
CREATE TABLE IF NOT EXISTS mirror_chatbridge (
repo TEXT PRIMARY KEY, -- the forge's name for the store
url TEXT NOT NULL, -- the upstream fossil's base URL
user TEXT NOT NULL, -- the bridge login on that fossil
secret TEXT NOT NULL, -- its password; never returned by any op
last_msgid INTEGER NOT NULL DEFAULT 0,
at TEXT NOT NULL DEFAULT '',-- when a message last arrived
last_err TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS mirror_chatbridge_msg (
repo TEXT NOT NULL,
msgid INTEGER NOT NULL, -- the upstream fossil's chat.msgid
seq INTEGER NOT NULL, -- the quarry channel sequence it became
PRIMARY KEY (repo, msgid)
);
CREATE INDEX IF NOT EXISTS chatbridge_msg_seq ON mirror_chatbridge_msg (repo, seq);`
// ChatBridge is one bridged channel as every reader but the poller
// sees it: no credential, ever.
type ChatBridge struct {
Repo string `json:"repo"`
URL string `json:"url"`
User string `json:"user"`
// LastMsgID is the upstream chat.msgid the poller has consumed
// through — the `name` parameter its next /chat-poll carries.
LastMsgID int64 `json:"lastMsgId"`
// At is when a message last arrived; LastErr the last failure, kept
// apart so "quiet" and "broken" are different answers.
At string `json:"at,omitempty"`
LastErr string `json:"lastErr,omitempty"`
}
// SetChatBridge configures (or reconfigures) a repository's bridge.
// Reconfiguring keeps the cursor when the URL is unchanged — an
// operator fixing a password has not made the messages already read
// unread — and resets it when the URL changes, because msgids belong
// to the fossil that minted them and replaying another server's
// numbering would import the wrong messages or none.
func (d *DB) SetChatBridge(repo, endpoint, user, secret string) error {
if repo == "" || user == "" {
return errors.New("oltp: chat bridge needs a repository and a user")
}
// http(s) only, the rule every URL quarry fetches on its own
// initiative lives under (see SetMirror).
u, err := url.Parse(endpoint)
if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
return fmt.Errorf("oltp: chat bridge url must be http or https, got %q", endpoint)
}
// Credentials belong in the user and secret columns, where the
// poller reads them and nothing else does. In the URL they would
// be copied into every log line, error string and status page
// that ever names the upstream.
if u.User != nil {
return errors.New("oltp: chat bridge url must not carry a user or password; " +
"give them as the user and password fields")
}
// A changed upstream invalidates the cursor AND the msgid map:
// msgids belong to the fossil that minted them, so a map carried
// across would let a new server's msgid tombstone a quarry message
// the old server's msgid had claimed. Same hazard DetachChatBridge
// handles, reached a different way.
prev, had, err := d.ChatBridgeFor(repo)
if err != nil {
return err
}
// One transaction: dropping the map and repointing the bridge are
// halves of one change. Committed separately, a failure between
// them leaves a map from the old server beside a row naming the
// new one — the exact pairing this is here to prevent.
tx, err := d.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
if had && prev.URL != endpoint {
if _, err := tx.Exec(`DELETE FROM mirror_chatbridge_msg WHERE repo = ?`, repo); err != nil {
return err
}
}
if _, err := tx.Exec(`
INSERT INTO mirror_chatbridge (repo, url, user, secret) VALUES (?, ?, ?, ?)
ON CONFLICT(repo) DO UPDATE SET
last_msgid = CASE WHEN url = excluded.url THEN last_msgid ELSE 0 END,
at = CASE WHEN url = excluded.url THEN at ELSE '' END,
url = excluded.url, user = excluded.user, secret = excluded.secret,
last_err = ''`,
repo, endpoint, user, secret); err != nil {
return err
}
return tx.Commit()
}
// DetachChatBridge forgets the bridge and the msgid map with it. The
// map is only meaningful against the cursor it was built beside, so
// keeping it would mean a re-attach could tombstone a quarry message
// on a number the new upstream reused for something else.
func (d *DB) DetachChatBridge(repo string) error {
if _, err := d.db.Exec(`DELETE FROM mirror_chatbridge WHERE repo = ?`, repo); err != nil {
return err
}
_, err := d.db.Exec(`DELETE FROM mirror_chatbridge_msg WHERE repo = ?`, repo)
return err
}
// ChatBridges lists every bridged repository, repo-ordered.
func (d *DB) ChatBridges() ([]ChatBridge, error) {
rows, err := d.db.Query(`SELECT repo, url, user, last_msgid, at, last_err
FROM mirror_chatbridge ORDER BY repo`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []ChatBridge
for rows.Next() {
var b ChatBridge
if err := rows.Scan(&b.Repo, &b.URL, &b.User, &b.LastMsgID, &b.At, &b.LastErr); err != nil {
return nil, err
}
out = append(out, b)
}
return out, rows.Err()
}
// ChatBridgeFor answers one repository's bridge, if it has one.
func (d *DB) ChatBridgeFor(repo string) (ChatBridge, bool, error) {
var b ChatBridge
err := d.db.QueryRow(`SELECT repo, url, user, last_msgid, at, last_err
FROM mirror_chatbridge WHERE repo = ?`, repo).
Scan(&b.Repo, &b.URL, &b.User, &b.LastMsgID, &b.At, &b.LastErr)
if errors.Is(err, sql.ErrNoRows) {
return ChatBridge{}, false, nil
}
return b, err == nil, err
}
// ChatBridgeSecret is the poller's own read of the credential. It is
// deliberately not a field on ChatBridge: a struct that carries a
// password ends up in a JSON response eventually.
func (d *DB) ChatBridgeSecret(repo string) (string, bool) {
var s string
err := d.db.QueryRow(`SELECT secret FROM mirror_chatbridge WHERE repo = ?`, repo).Scan(&s)
return s, err == nil
}
// SetChatBridgeCursor advances how far the poller has read and clears
// the error, which is what "a message arrived" means.
func (d *DB) SetChatBridgeCursor(repo string, msgid int64, at time.Time) error {
_, err := d.db.Exec(`UPDATE mirror_chatbridge
SET last_msgid = ?, at = ?, last_err = '' WHERE repo = ?`,
msgid, at.UTC().Format(time.RFC3339), repo)
return err
}
// RecordChatBridgeError remembers why the last poll failed. The cursor
// is untouched: a failed poll read nothing, and moving the cursor for
// it would skip the messages the next poll is meant to fetch.
func (d *DB) RecordChatBridgeError(repo string, runErr error) error {
msg := ""
if runErr != nil {
// The row is rendered on an admin page, so whatever a stdlib
// error decided to include in it is scrubbed here rather than
// trusted. net/http and net/url both put the whole URL in
// their messages, userinfo included.
msg = redactURLs(runErr.Error())
}
_, err := d.db.Exec(`UPDATE mirror_chatbridge SET last_err = ? WHERE repo = ?`, msg, repo)
return err
}
// MapChatBridgeMsg remembers which channel sequence an upstream msgid
// became, idempotently: a replayed poll must not double-post, and a
// later `mdel` naming that msgid has to find the message it deleted.
func (d *DB) MapChatBridgeMsg(repo string, msgid int64, seq uint64) error {
if repo == "" || msgid == 0 || seq == 0 {
return nil
}
_, err := d.db.Exec(`INSERT OR IGNORE INTO mirror_chatbridge_msg (repo, msgid, seq)
VALUES (?, ?, ?)`, repo, msgid, seq)
return err
}
// ChatBridgeSeq answers which sequence an upstream msgid became.
func (d *DB) ChatBridgeSeq(repo string, msgid int64) (uint64, bool) {
var seq uint64
err := d.db.QueryRow(`SELECT seq FROM mirror_chatbridge_msg WHERE repo = ? AND msgid = ?`,
repo, msgid).Scan(&seq)
return seq, err == nil
}
// PruneChatBridgeMsgs drops rows for messages the channel window no
// longer holds — the same rule the chat_ref rows live under, and for
// the same reason: the map describes the window, not the DAG.
func (d *DB) PruneChatBridgeMsgs(repo string, firstSeq uint64) error {
if repo == "" || firstSeq == 0 {
return nil
}
_, err := d.db.Exec(`DELETE FROM mirror_chatbridge_msg WHERE repo = ? AND seq < ?`,
repo, firstSeq)
return err
}
// urlUserinfo matches the `user:password@` of a URL inside free text.
//
// The userinfo run is greedy up to the LAST `@` before the path,
// because userinfo may itself contain one: a login like
// `bot@corp:hunter2@host` is ordinary in an org that uses email
// addresses as usernames, and a match that stopped at the first `@`
// would leave the password in the string. `/` still terminates the
// run, so a URL with no userinfo at all cannot swallow an `@` that
// appears later in the sentence.
var urlUserinfo = regexp.MustCompile(`([a-zA-Z][a-zA-Z0-9+.-]*://)[^/\s]*@`)
// redactURLs removes the credentials from any URL embedded in a
// message before it is stored. The bridge builds its own error strings
// carefully; this is for the ones it merely wraps, where a transport
// error carries the request URL and that URL came from configuration.
func redactURLs(msg string) string {
return urlUserinfo.ReplaceAllString(msg, "$1")
}