internal/store/mergequeue.go

 1package store
 2
 3import (
 4	"database/sql"
 5	"errors"
 6)
 7
 8// MRByID loads a merge request by its row id, without labels or review
 9// requests.
10func (s *Store) MRByID(id int64) (MR, error) {
11	m, err := scanMR(s.DB.QueryRow(mrSelect+" WHERE m.id = ?", id))
12	if errors.Is(err, sql.ErrNoRows) {
13		return m, ErrNotFound
14	}
15	return m, err
16}
17
18// QueueMerge queues a merge request to merge as userID with strategy
19// ("" for the default) once its gates pass, bound to the SSH key or API
20// token it was queued with (both 0 for a web session). Queueing again
21// replaces the queuer, strategy and credential and clears the reason.
22func (s *Store) QueueMerge(mrID, userID int64, strategy string, keyID, tokenID int64) error {
23	credential := ""
24	switch {
25	case keyID != 0:
26		credential = "key"
27	case tokenID != 0:
28		credential = "token"
29	}
30	_, err := s.DB.Exec(`
31		INSERT INTO mr_merge_queue (mr_id, user_id, strategy, credential, key_id, token_id)
32		VALUES (?, ?, ?, ?, ?, ?)
33		ON CONFLICT (mr_id) DO UPDATE SET user_id = excluded.user_id,
34			strategy = excluded.strategy, credential = excluded.credential,
35			key_id = excluded.key_id, token_id = excluded.token_id, reason = '',
36			queued_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')`,
37		mrID, userID, strategy, credential, nullID(keyID), nullID(tokenID))
38	return err
39}
40
41// QueueCredential is the credential a queued merge was queued with. Kind
42// is "key", "token", or "" for a web session; an id of 0 under "key" or
43// "token" means that credential has since been removed.
44type QueueCredential struct {
45	Kind    string
46	KeyID   int64
47	TokenID int64
48}
49
50func (s *Store) MergeQueueCredential(mrID int64) (QueueCredential, error) {
51	var q QueueCredential
52	err := s.DB.QueryRow(
53		"SELECT credential, COALESCE(key_id, 0), COALESCE(token_id, 0) FROM mr_merge_queue WHERE mr_id = ?",
54		mrID).Scan(&q.Kind, &q.KeyID, &q.TokenID)
55	if errors.Is(err, sql.ErrNoRows) {
56		return q, ErrNotFound
57	}
58	return q, err
59}
60
61// DequeueMerge takes a merge request off the queue, reporting whether it
62// was on it.
63func (s *Store) DequeueMerge(mrID int64) (bool, error) {
64	res, err := s.DB.Exec("DELETE FROM mr_merge_queue WHERE mr_id = ?", mrID)
65	if err != nil {
66		return false, err
67	}
68	n, err := res.RowsAffected()
69	return n > 0, err
70}
71
72// SetMergeQueueReason records why the last attempt at a queued merge did
73// not merge.
74func (s *Store) SetMergeQueueReason(mrID int64, reason string) error {
75	_, err := s.DB.Exec("UPDATE mr_merge_queue SET reason = ? WHERE mr_id = ?", reason, mrID)
76	return err
77}
78
79// QueuedMRsAtHead is the queued, open merge requests of a repository
80// whose head is sha: the ones a status reported on sha can move.
81func (s *Store) QueuedMRsAtHead(repoID int64, sha string) ([]int64, error) {
82	rows, err := s.DB.Query(`
83		SELECT m.id FROM mr_merge_queue q JOIN merge_requests m ON m.id = q.mr_id
84		WHERE m.repo_id = ? AND m.head_sha = ? AND m.state = 'open'
85		ORDER BY m.number`, repoID, sha)
86	if err != nil {
87		return nil, err
88	}
89	defer rows.Close()
90	var ids []int64
91	for rows.Next() {
92		var id int64
93		if err := rows.Scan(&id); err != nil {
94			return nil, err
95		}
96		ids = append(ids, id)
97	}
98	return ids, rows.Err()
99}