1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
// db.go stores the DM thread index in SQLite.
//
// The bot's own reply DMs are never kept in the outbox (see
// activitypub.Server.SendDM), so they are not retrievable at their AP id.
// Without a local record there is no way to tell what a reply like
// "@bot unboost" refers to. This file keeps a note id -> boosted URL index of
// both sides of every DM conversation so such replies can be resolved.
//
// The database lives at <data_dir>/booster-bot.db and is shared by all actors;
// rows carry the actor name so lookups stay scoped to one bot.
package main
import (
"database/sql"
"fmt"
"log"
"path/filepath"
"time"
_ "modernc.org/sqlite"
)
// Direction of a DM note, stored in dm_threads.direction.
const (
dmIncoming = "in" // a DM we received
dmOutgoing = "out" // a DM we sent
)
const migration001 = `-- DM thread index: maps DM note ids to the URL they are about.
--
-- One row per (note, URL) pair: a single DM may carry several URLs, and a
-- reply to it should be able to resolve all of them.
CREATE TABLE IF NOT EXISTS dm_threads (
note_id TEXT NOT NULL, -- AP id of a Note in the conversation
target_url TEXT NOT NULL, -- resolved AP id of the boosted object
actor_name TEXT NOT NULL, -- which of our actors the DM belongs to
peer TEXT NOT NULL, -- actor URL of the human on the other side
direction TEXT NOT NULL, -- 'in' (received) or 'out' (sent)
created_at INTEGER NOT NULL, -- Unix timestamp
PRIMARY KEY (note_id, target_url)
);
CREATE INDEX IF NOT EXISTS idx_dm_threads_actor ON dm_threads(actor_name);
`
// openDB opens (creating if needed) the booster-bot database in dataDir and
// applies any pending migrations.
func openDB(dataDir string) (*sql.DB, error) {
path := filepath.Join(dataDir, "booster-bot.db")
// Pragmas travel in the DSN so they apply to every pooled connection:
// busy_timeout makes a writer wait for the lock instead of failing with
// SQLITE_BUSY, and WAL keeps readers from blocking the writer. A post-open
// "PRAGMA …" would only configure whichever pooled connection ran it.
db, err := sql.Open("sqlite",
"file:"+path+"?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)")
if err != nil {
return nil, fmt.Errorf("open database %s: %w", path, err)
}
if err := runMigrations(db); err != nil {
db.Close()
return nil, fmt.Errorf("migrate database %s: %w", path, err)
}
log.Printf("db: opened %s", path)
return db, nil
}
// runMigrations applies all migrations that have not been applied yet.
func runMigrations(db *sql.DB) error {
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_version (
version INTEGER PRIMARY KEY,
applied_at INTEGER NOT NULL
)`); err != nil {
return fmt.Errorf("create schema_version: %w", err)
}
migrations := []struct {
version int
sql string
}{
{1, migration001},
}
for _, m := range migrations {
var applied int
if err := db.QueryRow(
"SELECT COUNT(*) FROM schema_version WHERE version = ?", m.version,
).Scan(&applied); err != nil {
return fmt.Errorf("check migration %d: %w", m.version, err)
}
if applied > 0 {
continue
}
tx, err := db.Begin()
if err != nil {
return fmt.Errorf("begin migration %d: %w", m.version, err)
}
if _, err := tx.Exec(m.sql); err != nil {
tx.Rollback()
return fmt.Errorf("apply migration %d: %w", m.version, err)
}
if _, err := tx.Exec(
"INSERT INTO schema_version (version, applied_at) VALUES (?, ?)",
m.version, time.Now().Unix(),
); err != nil {
tx.Rollback()
return fmt.Errorf("record migration %d: %w", m.version, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit migration %d: %w", m.version, err)
}
log.Printf("db: applied migration %d", m.version)
}
return nil
}
// recordDMThread remembers that the DM note noteID is about targetURL, so that
// later replies to it can be resolved back to that URL.
//
// Failures are logged, not returned: losing a thread record degrades unboost
// by reply, but must never abort the boost it belongs to. A repeated
// (noteID, targetURL) pair is ignored rather than treated as an error, since
// the same URL may legitimately appear twice in one DM.
func recordDMThread(db *sql.DB, noteID, targetURL, actorName, peer, direction string) {
if db == nil || noteID == "" || targetURL == "" {
return
}
if _, err := db.Exec(`INSERT OR IGNORE INTO dm_threads
(note_id, target_url, actor_name, peer, direction, created_at)
VALUES (?, ?, ?, ?, ?, ?)`,
noteID, targetURL, actorName, peer, direction, time.Now().Unix(),
); err != nil {
log.Printf("db: record DM thread %s -> %s: %v", noteID, targetURL, err)
}
}
// lookupDMThread returns the URLs that the DM note noteID is about, for the
// given actor. Returns nil if the note is unknown — which is the normal case
// for conversations that started before this index existed.
func lookupDMThread(db *sql.DB, noteID, actorName string) []string {
if db == nil || noteID == "" {
return nil
}
rows, err := db.Query(
`SELECT target_url FROM dm_threads
WHERE note_id = ? AND actor_name = ?
ORDER BY rowid`,
noteID, actorName)
if err != nil {
log.Printf("db: look up DM thread %s: %v", noteID, err)
return nil
}
defer rows.Close()
var urls []string
for rows.Next() {
var u string
if err := rows.Scan(&u); err != nil {
log.Printf("db: scan DM thread %s: %v", noteID, err)
return nil
}
urls = append(urls, u)
}
if err := rows.Err(); err != nil {
log.Printf("db: iterate DM thread %s: %v", noteID, err)
return nil
}
return urls
}
|