// attn.go holds the forge-local state RFC 0011 §5's attention stream
// needs and the stream itself cannot carry: how far each account has
// read (attn_read), the per-account preferences a person set for
// themselves (account_pref), and which events a repository's activity
// robot speaks aloud in its channel (chat_robot, §10).
//
// All three are social state, not truth: wiping them loses a read
// pointer, a checkbox and a robot's subscription, never a fact the DAG
// holds (RFC 0001 §11).
package oltp
import "time"
const attnSchema = `
CREATE TABLE IF NOT EXISTS attn_read (
account TEXT NOT NULL PRIMARY KEY,
seq INTEGER NOT NULL, -- highest attention seq the account has seen
at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS account_pref (
account TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT NOT NULL,
PRIMARY KEY (account, key)
);
CREATE TABLE IF NOT EXISTS chat_robot (
repo TEXT NOT NULL PRIMARY KEY,
events TEXT NOT NULL -- webhook pattern syntax; empty means off
);
CREATE TABLE IF NOT EXISTS chat_robot_line (
code TEXT NOT NULL, -- the channel's project code, as chat keys it
seq INTEGER NOT NULL, -- the robot line's JetStream stream sequence
at TEXT NOT NULL,
PRIMARY KEY (code, seq)
);
-- The (code, seq) primary key IS this index; the separate one was a
-- second copy of the same b-tree, dropped here so an existing forge
-- sheds it too.
DROP INDEX IF EXISTS chat_robot_line_seq;`
// robotLineKeep is how long a robot line is remembered: exactly the
// chat window, because a row that outlives its message describes a
// message the broker has stopped counting — and chat.Unread would then
// subtract it from an unread run that no longer contains it.
const robotLineKeep = 7 * 24 * time.Hour
// MarkAttentionRead advances an account's attention read pointer, and
// only ever forwards. The pointer is the local unread count's floor —
// "events after this" — so a stale client replaying an older cursor
// must not resurrect mail the person has already dismissed.
func (d *DB) MarkAttentionRead(account string, seq uint64) error {
if account == "" {
return nil
}
_, err := d.w.Exec(`INSERT INTO attn_read (account, seq, at)
VALUES (?, ?, ?)
ON CONFLICT (account) DO UPDATE SET seq = excluded.seq, at = excluded.at
WHERE attn_read.seq < excluded.seq`,
account, seq, time.Now().UTC().Format(time.RFC3339Nano))
return err
}
// AttentionReadSeq answers the account's pointer — 0 for someone who
// has never marked anything, which reads as "everything in the window
// is unread", the correct answer for an inbox nobody has opened.
func (d *DB) AttentionReadSeq(account string) uint64 {
if account == "" {
return 0
}
var seq uint64
d.db.QueryRow(`SELECT seq FROM attn_read WHERE account = ?`, account).Scan(&seq)
return seq
}
// SetAccountPref stores one preference for one account; an empty value
// deletes the row, so "off" is the absence of a preference rather than
// a second spelling of it.
func (d *DB) SetAccountPref(account, key, value string) error {
if account == "" || key == "" {
return nil
}
if value == "" {
_, err := d.w.Exec(`DELETE FROM account_pref WHERE account = ? AND key = ?`, account, key)
return err
}
_, err := d.w.Exec(`INSERT INTO account_pref (account, key, value)
VALUES (?, ?, ?)
ON CONFLICT (account, key) DO UPDATE SET value = excluded.value`,
account, key, value)
return err
}
// AccountPref reads one preference, empty when unset.
func (d *DB) AccountPref(account, key string) string {
if account == "" || key == "" {
return ""
}
var v string
d.db.QueryRow(`SELECT value FROM account_pref WHERE account = ? AND key = ?`,
account, key).Scan(&v)
return v
}
// SetChatRobot stores a repository's robot event patterns (webhook
// pattern syntax). Empty turns the robot off by deleting the row: a
// repository with no row and a repository with an empty pattern must
// not be two different states for the consumer to tell apart.
func (d *DB) SetChatRobot(repo, events string) error {
if repo == "" {
return nil
}
if events == "" {
_, err := d.w.Exec(`DELETE FROM chat_robot WHERE repo = ?`, repo)
return err
}
_, err := d.w.Exec(`INSERT INTO chat_robot (repo, events) VALUES (?, ?)
ON CONFLICT (repo) DO UPDATE SET events = excluded.events`, repo, events)
return err
}
// ChatRobot answers one repository's patterns, empty when the robot is
// off.
func (d *DB) ChatRobot(repo string) string {
var v string
d.db.QueryRow(`SELECT events FROM chat_robot WHERE repo = ?`, repo).Scan(&v)
return v
}
// ChatRobots answers every opted-in repository's patterns — what the
// robot consumer reads once per event instead of once per repository.
func (d *DB) ChatRobots() (map[string]string, error) {
out := map[string]string{}
rows, err := d.db.Query(`SELECT repo, events FROM chat_robot`)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var repo, events string
if err := rows.Scan(&repo, &events); err != nil {
return nil, err
}
out[repo] = events
}
return out, rows.Err()
}
// RecordRobotLine notes that one of a channel's sequences is the
// activity robot's own line (RFC 0011 §10) — the fact chat.Unread
// subtracts so a badge never lights up for the robot, and never has to
// read the window to find out.
//
// Idempotent on (code, seq): a redelivered event says the same thing,
// and a replay must not double the subtraction.
func (d *DB) RecordRobotLine(code string, seq uint64, at string) error {
if code == "" || seq == 0 {
return nil
}
if at == "" {
at = time.Now().UTC().Format(time.RFC3339Nano)
}
if _, err := d.w.Exec(`INSERT OR IGNORE INTO chat_robot_line (code, seq, at)
VALUES (?, ?, ?)`, code, seq, at); err != nil {
return err
}
// Swept here rather than on a timer, for PruneChatRefs' reason: the
// window only moves when something is posted. By AGE, not by
// sequence — sequences belong to the whole stream, so "seq minus
// the per-channel cap" is not a point this channel's window has
// passed.
// A TEXT comparison, which is what RFC3339Nano is for: fixed-width
// UTC stamps sort lexically the way they sort chronologically.
_, err := d.w.Exec(`DELETE FROM chat_robot_line WHERE code = ? AND at < ?`,
code, time.Now().UTC().Add(-robotLineKeep).Format(time.RFC3339Nano))
return err
}
// PurgeRobotLines forgets a channel's robot lines, for when the channel
// itself is purged with its repository.
func (d *DB) PurgeRobotLines(code string) error {
_, err := d.w.Exec(`DELETE FROM chat_robot_line WHERE code = ?`, code)
return err
}
// RobotLines counts a channel's robot lines after since — one bounded
// index range scan (count(*) walks the rows it matches, so the caller
// bounds since by the window), and the only thing chat.Unread needs to
// turn the broker's pending count into an unread count.
func (d *DB) RobotLines(code string, since uint64) (int, error) {
var n int
err := d.db.QueryRow(`SELECT count(*) FROM chat_robot_line
WHERE code = ? AND seq > ?`, code, since).Scan(&n)
return n, err
}