Sign in

Search issue titles and bodies, change requests and comments. Private repositories you cannot read are not included.
vole
Clone

Read only, no credentials:

https://rodent.software/vole.git

Write over SSH with a registered key:

git clone ssh://git@rodent.software:2222/vole.git

A free and open-source forge that lives in your Git repo.

Code

396 commit(s) on main

Compare History
Turn off the disk reserve for test-spawned runners Keenan Nemetz · 15m ago c284cb2c
history blame raw 16.1 KiB
1package index
2
3// The activity feed, materialised.
4//
5// The feed is a stream of events: an issue opened, a comment, an approval, a
6// closed or merged change, a release. The old page assembled those from the
7// source rows on every request and sorted them, so a page cost the repository's
8// whole history. Here they are rows with their own time and id, indexed by
9// (repository, time), so a page is an index range read and an unfiltered total
10// is a stored count.
11//
12// The row carries the title and assignees the feed shows rather than joining the
13// entity at read time: the feed has always shown the current title, and the
14// entity's events are replaced when the entity changes. URLs are built from ids
15// and the repository name at render time, so a rename cannot leave a dead link.
16
17import (
18 "database/sql"
19 "fmt"
20 "strings"
21 "time"
22
23 "rodent.software/vole/core/git"
24 "rodent.software/vole/forge/repo"
25 "rodent.software/vole/metadata"
26)
27
28// EventRow is one feed event. The renderer turns it into a row: the ids are what
29// a URL is made of, and the state is what turns a state event into "closed" or
30// "reopened".
31type EventRow struct {
32 RepoID string
33 ID string
34 Kind string // issue | change | release
35 Type string // open | comment | approval | state | release
36 At int64
37 Actor string
38 Title string
39 Assignees string // \x1f-joined, issue events only
40 EntityID string
41 ParentID string
42 CommentID string
43 State string
44}
45
46// insertEvent writes one row.
47func insertEvent(tx execer, repoID string, e EventRow) error {
48 _, err := tx.Exec(
49 `INSERT INTO events(repo_id, id, kind, type, at, actor, title, assignees,
50 entity_id, parent_id, comment_id, state)
51 VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
52 repoID, e.ID, e.Kind, e.Type, e.At, e.Actor, e.Title, e.Assignees,
53 e.EntityID, e.ParentID, e.CommentID, e.State)
54 return err
55}
56
57// deleteEventsForEntity removes an entity's events and reports how many rows
58// went, so the repository's stored count can be adjusted by the same amount.
59func deleteEventsForEntity(tx execer, repoID, kind, id string) (int, error) {
60 res, err := tx.Exec(`DELETE FROM events WHERE repo_id=? AND kind=? AND entity_id=?`, repoID, kind, id)
61 if err != nil {
62 return 0, err
63 }
64 n, err := res.RowsAffected()
65 if err != nil {
66 return 0, err
67 }
68 return int(n), nil
69}
70
71// insertIssueEvents replaces one issue's events: the opening, every comment, and
72// every state change the branch recorded. Deleted comments keep their tombstone
73// event, as the feed has always shown that something was said.
74func insertIssueEvents(tx *sql.Tx, repoID string, iss *metadata.Issue) (inserted, deleted int, err error) {
75 deleted, err = deleteEventsForEntity(tx, repoID, "issue", iss.ID)
76 if err != nil {
77 return 0, deleted, err
78 }
79 assignees := strings.Join(iss.Assignees, "\x1f")
80 base := EventRow{
81 Kind: "issue", At: iss.Created.Unix(), Actor: iss.Author,
82 Title: iss.Title, Assignees: assignees, EntityID: iss.ID,
83 }
84 open := base
85 open.ID = "issue:" + iss.ID + ":open"
86 open.Type = "open"
87 if err := insertEvent(tx, repoID, open); err != nil {
88 return 0, deleted, err
89 }
90 inserted++
91 for _, c := range iss.Comments {
92 ev := base
93 ev.ID = "issue:" + iss.ID + ":comment:" + c.ID
94 ev.Type = "comment"
95 ev.At = c.Created.Unix()
96 ev.Actor = c.Author
97 ev.ParentID = iss.ID
98 ev.CommentID = c.ID
99 if err := insertEvent(tx, repoID, ev); err != nil {
100 return inserted, deleted, err
101 }
102 inserted++
103 }
104 states, err := entityStateEvents(tx, repoID, "issue", iss.ID)
105 if err != nil {
106 return inserted, deleted, err
107 }
108 for _, ev := range states {
109 ev.Kind = "issue"
110 ev.Title = iss.Title
111 ev.Assignees = assignees
112 if err := insertEvent(tx, repoID, ev); err != nil {
113 return inserted, deleted, err
114 }
115 inserted++
116 }
117 return inserted, deleted, nil
118}
119
120// insertRequestEvents is the same for a change: opening, comments, approvals and
121// state changes.
122func insertRequestEvents(tx *sql.Tx, repoID string, req *metadata.Request) (inserted, deleted int, err error) {
123 deleted, err = deleteEventsForEntity(tx, repoID, "change", req.ID)
124 if err != nil {
125 return 0, deleted, err
126 }
127 base := EventRow{
128 Kind: "change", At: req.Created.Unix(), Actor: req.Author,
129 Title: req.Title, EntityID: req.ID,
130 }
131 open := base
132 open.ID = "request:" + req.ID + ":open"
133 open.Type = "open"
134 if err := insertEvent(tx, repoID, open); err != nil {
135 return 0, deleted, err
136 }
137 inserted++
138 for _, c := range req.Comments {
139 ev := base
140 ev.ID = "request:" + req.ID + ":comment:" + c.ID
141 ev.Type = "comment"
142 ev.At = c.Created.Unix()
143 ev.Actor = c.Author
144 ev.ParentID = req.ID
145 ev.CommentID = c.ID
146 if err := insertEvent(tx, repoID, ev); err != nil {
147 return inserted, deleted, err
148 }
149 inserted++
150 }
151 for _, a := range req.Approvals {
152 ev := base
153 ev.ID = "request:" + req.ID + ":approval:" + a.Principal
154 ev.Type = "approval"
155 ev.At = a.Created.Unix()
156 ev.Actor = a.Principal
157 if err := insertEvent(tx, repoID, ev); err != nil {
158 return inserted, deleted, err
159 }
160 inserted++
161 }
162 states, err := entityStateEvents(tx, repoID, "request", req.ID)
163 if err != nil {
164 return inserted, deleted, err
165 }
166 for _, ev := range states {
167 ev.Kind = "change"
168 ev.Title = req.Title
169 if err := insertEvent(tx, repoID, ev); err != nil {
170 return inserted, deleted, err
171 }
172 inserted++
173 }
174 return inserted, deleted, nil
175}
176
177// entityStateEvents reads the state changes for one entity. The ordinal is part
178// of the event id so two changes in the same second keep the branch's order.
179func entityStateEvents(tx *sql.Tx, repoID, kind, id string) ([]EventRow, error) {
180 rows, err := tx.Query(
181 `SELECT sha, state, actor, at, ordinal FROM state_changes
182 WHERE repo_id=? AND kind=? AND id=? ORDER BY ordinal`, repoID, kind, id)
183 if err != nil {
184 return nil, err
185 }
186 defer rows.Close()
187 var out []EventRow
188 for rows.Next() {
189 var (
190 sha, state, actor string
191 at int64
192 ordinal int
193 )
194 if err := rows.Scan(&sha, &state, &actor, &at, &ordinal); err != nil {
195 return nil, err
196 }
197 prefix := "issue"
198 if kind == "request" {
199 prefix = "request"
200 }
201 out = append(out, EventRow{
202 ID: fmt.Sprintf("%s:%s:state:%08d", prefix, id, ordinal), Type: "state",
203 At: at, Actor: actor, EntityID: id, State: state,
204 })
205 }
206 return out, rows.Err()
207}
208
209// replaceReleaseEvents replaces a repository's release events with the tags that
210// exist now.
211func replaceReleaseEvents(tx *sql.Tx, repoID string, tags []git.Tag) (inserted, deleted int, err error) {
212 res, err := tx.Exec(`DELETE FROM events WHERE repo_id=? AND kind='release'`, repoID)
213 if err != nil {
214 return 0, 0, err
215 }
216 n, err := res.RowsAffected()
217 if err != nil {
218 return 0, 0, err
219 }
220 deleted = int(n)
221 for _, t := range tags {
222 ev := EventRow{
223 ID: "release:" + t.Name, Kind: "release", Type: "release",
224 At: t.When, Actor: t.Tagger, Title: t.Name, EntityID: t.Name,
225 }
226 if err := insertEvent(tx, repoID, ev); err != nil {
227 return inserted, deleted, err
228 }
229 inserted++
230 }
231 return inserted, deleted, nil
232}
233
234// bumpEventCount keeps the stored total in step with the rows a refresh added or
235// removed.
236func bumpEventCount(tx execer, repoID string, delta int) error {
237 if delta == 0 {
238 return nil
239 }
240 _, err := tx.Exec(`UPDATE repos SET events_count = MAX(0, events_count + ?) WHERE id=?`, delta, repoID)
241 return err
242}
243
244// RefreshTags replaces a repository's release events from its tags. It is called
245// after any push (post-receive) and after a release written in the browser,
246// because a tag is a ref and no metadata write announces it.
247func (s *Store) RefreshTags(m *repo.Meta) error {
248 tags, err := m.Git().Tags()
249 if err != nil {
250 return err
251 }
252 tx, err := s.db.Begin()
253 if err != nil {
254 return err
255 }
256 defer tx.Rollback()
257 inserted, deleted, err := replaceReleaseEvents(tx, m.ID, tags)
258 if err != nil {
259 return err
260 }
261 if err := replaceTagRows(tx, m.ID, tags); err != nil {
262 return err
263 }
264 if err := bumpEventCount(tx, m.ID, inserted-deleted); err != nil {
265 return err
266 }
267 return tx.Commit()
268}
269
270// replaceTagRows stores the tag list the releases sections read. A tag is a ref,
271// so this is not a metadata commit either: every push refreshes it alongside the
272// release events, and a rebuild does the same.
273func replaceTagRows(tx execer, repoID string, tags []git.Tag) error {
274 if _, err := tx.Exec(`DELETE FROM repo_tags WHERE repo_id=?`, repoID); err != nil {
275 return err
276 }
277 for _, t := range tags {
278 annotated := 0
279 if t.Annotated {
280 annotated = 1
281 }
282 if _, err := tx.Exec(
283 `INSERT INTO repo_tags(repo_id, name, oid, commit_sha, annotated, tagger, when_at, subject)
284 VALUES(?,?,?,?,?,?,?,?)`,
285 repoID, t.Name, t.OID, t.Commit, annotated, t.Tagger, t.When, t.Subject); err != nil {
286 return err
287 }
288 }
289 return nil
290}
291
292// eventQuery is the set of narrowings the feed, a profile and the directory
293// share. kind is the event kind (issue, change, release); typ is what happened
294// (open, comment, approval, state, release). Either may be empty.
295type eventQuery struct {
296 kind string
297 typ string
298 actor string
299 assignee string
300 // notKind excludes one kind. A profile uses it to leave releases out: a
301 // release event names the tag's tagger, which is a Git identity and not the
302 // credential that proves an account, so it is not somebody's work here.
303 notKind string
304}
305
306// where is the WHERE clause the count and the page share.
307func (q eventQuery) where(repoIDs []string) (string, []any) {
308 conds := []string{`repo_id IN (` + questionMarks(len(repoIDs)) + `)`}
309 args := make([]any, 0, len(repoIDs)+3)
310 for _, id := range repoIDs {
311 args = append(args, id)
312 }
313 if q.kind != "" {
314 conds = append(conds, `kind = ?`)
315 args = append(args, q.kind)
316 }
317 if q.typ != "" {
318 conds = append(conds, `type = ?`)
319 args = append(args, q.typ)
320 }
321 if q.notKind != "" {
322 conds = append(conds, `kind != ?`)
323 args = append(args, q.notKind)
324 }
325 if q.actor != "" {
326 conds = append(conds, `actor = ?`)
327 args = append(args, q.actor)
328 }
329 if q.assignee != "" {
330 // Only issue events carry assignees. The column is a \x1f-joined list;
331 // char(31) keeps the delimiter out of the query text.
332 conds = append(conds,
333 `kind = 'issue'`,
334 `instr(char(31)||assignees||char(31), char(31)||?||char(31)) > 0`)
335 args = append(args, q.assignee)
336 }
337 return strings.Join(conds, " AND "), args
338}
339
340// unfiltered reports whether the query is the whole feed, which is the one case
341// a stored per-repository count can answer.
342func (q eventQuery) unfiltered() bool {
343 return q.kind == "" && q.typ == "" && q.actor == "" && q.assignee == "" && q.notKind == ""
344}
345
346// EventCount is how many events the filters select. The unfiltered total is the
347// stored per-repository count, so a pager does not count a large repository's
348// history on every page; a filtered total is an index scan without the sort.
349func (s *Store) EventCount(repoIDs []string, kind, actor, assignee string) (int, error) {
350 return s.countEvents(repoIDs, eventQuery{kind: kind, actor: actor, assignee: assignee})
351}
352
353func (s *Store) countEvents(repoIDs []string, q eventQuery) (int, error) {
354 if len(repoIDs) == 0 {
355 return 0, nil
356 }
357 if q.unfiltered() {
358 args := make([]any, 0, len(repoIDs))
359 for _, id := range repoIDs {
360 args = append(args, id)
361 }
362 var n int
363 err := s.db.QueryRow(
364 `SELECT COALESCE(SUM(events_count),0) FROM repos WHERE id IN (`+questionMarks(len(repoIDs))+`)`,
365 args...).Scan(&n)
366 return n, err
367 }
368 where, args := q.where(repoIDs)
369 var n int
370 err := s.db.QueryRow(`SELECT COUNT(*) FROM events WHERE `+where, args...).Scan(&n)
371 return n, err
372}
373
374// eventColumns is the row shape every event query returns.
375const eventColumns = `repo_id, id, kind, type, at, actor, title, assignees,
376 entity_id, parent_id, comment_id, state`
377
378// eventPageQuery builds the query for one page.
379//
380// The index narrows first. One repository is a range of the (repository, time)
381// index; one account is a range of the actor index, which is what makes a profile
382// a read of that person's rows rather than a scan of everything they can see; a
383// whole-instance feed walks the global time index. In every case the page stops
384// after offset+limit matches, so no query sorts a repository's history.
385func eventPageQuery(repoIDs []string, q eventQuery, limit, offset int) (string, []any) {
386 where, args := q.where(repoIDs)
387 args = append(args, limit, offset)
388 switch {
389 case len(repoIDs) == 1 && q.actor == "":
390 return `SELECT ` + eventColumns + ` FROM events WHERE ` + where +
391 ` ORDER BY at DESC, id DESC LIMIT ? OFFSET ?`, args
392 case q.actor != "":
393 return `SELECT ` + eventColumns + ` FROM events INDEXED BY events_actor WHERE ` + where +
394 ` ORDER BY at DESC, id DESC LIMIT ? OFFSET ?`, args
395 }
396 return `SELECT ` + eventColumns + ` FROM events INDEXED BY events_time WHERE ` + where +
397 ` ORDER BY at DESC, id DESC LIMIT ? OFFSET ?`, args
398}
399
400// Events reads one page of the feed, newest first.
401func (s *Store) Events(repoIDs []string, kind, actor, assignee string, limit, offset int) ([]EventRow, error) {
402 return s.eventPage(repoIDs, eventQuery{kind: kind, actor: actor, assignee: assignee}, limit, offset)
403}
404
405// ActorEvents reads one page of one account's events, optionally narrowed to a
406// kind and a type. A profile asks for "alice's reviews" by naming the approval
407// type; the repositories are the ones the viewer can read, so the page never
408// counts a repository they cannot open. Releases are left out: see eventQuery.
409func (s *Store) ActorEvents(repoIDs []string, kind, typ, actor string, limit, offset int) ([]EventRow, error) {
410 return s.eventPage(repoIDs, eventQuery{kind: kind, typ: typ, actor: actor, notKind: "release"}, limit, offset)
411}
412
413// ActorEventCount is the total ActorEvents pages through.
414func (s *Store) ActorEventCount(repoIDs []string, kind, typ, actor string) (int, error) {
415 return s.countEvents(repoIDs, eventQuery{kind: kind, typ: typ, actor: actor, notKind: "release"})
416}
417
418func (s *Store) eventPage(repoIDs []string, q eventQuery, limit, offset int) ([]EventRow, error) {
419 if len(repoIDs) == 0 || limit <= 0 {
420 return nil, nil
421 }
422 query, args := eventPageQuery(repoIDs, q, limit, offset)
423 rows, err := s.db.Query(query, args...)
424 if err != nil {
425 return nil, err
426 }
427 defer rows.Close()
428 var out []EventRow
429 for rows.Next() {
430 var r EventRow
431 if err := rows.Scan(&r.RepoID, &r.ID, &r.Kind, &r.Type, &r.At, &r.Actor,
432 &r.Title, &r.Assignees, &r.EntityID, &r.ParentID, &r.CommentID, &r.State); err != nil {
433 return nil, err
434 }
435 out = append(out, r)
436 }
437 return out, rows.Err()
438}
439
440// questionMarks renders "?,?,?" for an IN clause.
441func questionMarks(n int) string {
442 if n <= 0 {
443 return ""
444 }
445 return strings.TrimSuffix(strings.Repeat("?,", n), ",")
446}
447
448// LastEventTimes returns the newest event time per repository. It is the card's
449// "updated" time, read for a whole page of cards in one grouped query rather
450// than one query per card.
451func (s *Store) LastEventTimes(repoIDs []string) (map[string]time.Time, error) {
452 if len(repoIDs) == 0 {
453 return nil, nil
454 }
455 args := make([]any, 0, len(repoIDs))
456 for _, id := range repoIDs {
457 args = append(args, id)
458 }
459 rows, err := s.db.Query(
460 `SELECT repo_id, MAX(at) FROM events WHERE repo_id IN (`+questionMarks(len(repoIDs))+`)
461 GROUP BY repo_id`, args...)
462 if err != nil {
463 return nil, err
464 }
465 defer rows.Close()
466 out := map[string]time.Time{}
467 for rows.Next() {
468 var id string
469 var at int64
470 if err := rows.Scan(&id, &at); err != nil {
471 return nil, err
472 }
473 out[id] = time.Unix(at, 0).UTC()
474 }
475 return out, rows.Err()
476}
477
478// LastReleases returns the newest release event's tag name per repository, for a
479// card's release chip. It reads the same rows the feed does rather than listing
480// tags from git again.
481func (s *Store) LastReleases(repoIDs []string) (map[string]string, error) {
482 if len(repoIDs) == 0 {
483 return nil, nil
484 }
485 args := make([]any, 0, len(repoIDs))
486 for _, id := range repoIDs {
487 args = append(args, id)
488 }
489 rows, err := s.db.Query(
490 `SELECT repo_id, title FROM events
491 WHERE kind='release' AND repo_id IN (`+questionMarks(len(repoIDs))+`)
492 ORDER BY at DESC, id DESC`, args...)
493 if err != nil {
494 return nil, err
495 }
496 defer rows.Close()
497 out := map[string]string{}
498 for rows.Next() {
499 var repoID, name string
500 if err := rows.Scan(&repoID, &name); err != nil {
501 return nil, err
502 }
503 if _, ok := out[repoID]; !ok {
504 out[repoID] = name
505 }
506 }
507 return out, rows.Err()
508}
509