Skip to content
quarry Artifact [c9d255a994]
Login

Artifact [c9d255a994]

Artifact c9d255a994a54210cea046d4de233712de9c8fecfdd34e1db9aaa0bfaefd77c8:


// 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)
);
CREATE INDEX IF NOT EXISTS chat_robot_line_seq ON chat_robot_line (code, seq);`

// robotLineKeep is how long a robot line is remembered: the chat
// window is seven days, so a row a day past it is certainly describing
// a message the channel no longer holds.
const robotLineKeep = 8 * 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.db.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.db.Exec(`DELETE FROM account_pref WHERE account = ? AND key = ?`, account, key)
		return err
	}
	_, err := d.db.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.db.Exec(`DELETE FROM chat_robot WHERE repo = ?`, repo)
		return err
	}
	_, err := d.db.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.db.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.
	_, err := d.db.Exec(`DELETE FROM chat_robot_line WHERE code = ? AND at < ?`,
		code, time.Now().UTC().Add(-robotLineKeep).Format(time.RFC3339Nano))
	return err
}

// RobotLines counts a channel's robot lines after since — O(1) against
// the index, 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
}