alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "time"
8
9 "github.com/bluesky-social/indigo/atproto/syntax"
10)
11
12type GitRepoMigrationStatus string
13
14const (
15 GitRepoMigrationStatusPending GitRepoMigrationStatus = "pending"
16 GitRepoMigrationStatusRunning GitRepoMigrationStatus = "running"
17 GitRepoMigrationStatusDone GitRepoMigrationStatus = "done"
18 GitRepoMigrationStatusFailed GitRepoMigrationStatus = "failed"
19)
20
21const (
22 GitRepoMigrationSourceGitHub = "github"
23)
24
25type GitRepoMigrations []GitRepoMigration
26
27func (m GitRepoMigrations) AnyActive() bool {
28 for _, r := range m {
29 if r.Status == GitRepoMigrationStatusPending || r.Status == GitRepoMigrationStatusRunning {
30 return true
31 }
32 }
33 return false
34}
35
36type GitRepoMigration struct {
37 ID int64
38 OwnerDid syntax.DID
39 SourceKind string
40 CloneUrl string
41 Name string
42 Knot string
43 Description string
44 Website string
45 Topics []string
46 SessionID string
47 Status GitRepoMigrationStatus
48 ErrorMsg string
49 UpdatedAt time.Time
50}
51
52func InsertGitRepoMigrations(ctx context.Context, e *DB, rows []GitRepoMigration) error {
53 if len(rows) == 0 {
54 return nil
55 }
56 txx, err := e.BeginTx(ctx, nil)
57 if err != nil {
58 return err
59 }
60 defer txx.Rollback()
61
62 stmt, err := txx.PrepareContext(ctx, `
63 insert into gitrepo_migrations
64 (owner_did, source_kind, clone_url, name, knot, description, session_id,
65 status, error_msg, updated_at)
66 values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
67 on conflict(owner_did, name) do update set
68 source_kind = excluded.source_kind,
69 clone_url = excluded.clone_url,
70 knot = excluded.knot,
71 description = excluded.description,
72 session_id = excluded.session_id,
73 status = 'pending',
74 error_msg = '',
75 updated_at = excluded.updated_at
76 `)
77 if err != nil {
78 return err
79 }
80 defer stmt.Close()
81
82 now := time.Now().UTC().Format(time.RFC3339)
83 for _, r := range rows {
84 status := r.Status
85 if status == "" {
86 status = GitRepoMigrationStatusPending
87 }
88 if _, err := stmt.ExecContext(ctx,
89 r.OwnerDid, r.SourceKind, r.CloneUrl, r.Name, r.Knot, r.Description, r.SessionID,
90 status, r.ErrorMsg, now,
91 ); err != nil {
92 return err
93 }
94 }
95 return txx.Commit()
96}
97
98func ClaimNextPending(ctx context.Context, e Execer) (*GitRepoMigration, bool, error) {
99 var (
100 m GitRepoMigration
101 updatedAt string
102 )
103 err := e.QueryRowContext(ctx, `
104 update gitrepo_migrations
105 set status = 'running',
106 updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
107 where id = (
108 select id from gitrepo_migrations gm
109 where status = 'pending'
110 and not exists (
111 select 1 from gitrepo_migrations gm2
112 where gm2.owner_did = gm.owner_did
113 and gm2.status = 'running'
114 )
115 order by id
116 limit 1
117 )
118 returning id, owner_did, source_kind, clone_url, name, knot,
119 description, session_id, status, error_msg, updated_at
120 `).Scan(
121 &m.ID, &m.OwnerDid, &m.SourceKind, &m.CloneUrl, &m.Name, &m.Knot,
122 &m.Description, &m.SessionID, &m.Status, &m.ErrorMsg,
123 &updatedAt,
124 )
125 if errors.Is(err, sql.ErrNoRows) {
126 return nil, false, nil
127 }
128 if err != nil {
129 return nil, false, err
130 }
131 if t, err := time.Parse(time.RFC3339, updatedAt); err == nil {
132 m.UpdatedAt = t
133 }
134 return &m, true, nil
135}
136
137func MarkGitRepoMigrationDone(ctx context.Context, e Execer, id int64) error {
138 _, err := e.ExecContext(ctx, `
139 update gitrepo_migrations
140 set status = 'done',
141 error_msg = '',
142 updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
143 where id = ?
144 `, id)
145 return err
146}
147
148func MarkGitRepoMigrationFailed(ctx context.Context, e Execer, id int64, errMsg string) error {
149 _, err := e.ExecContext(ctx, `
150 update gitrepo_migrations
151 set status = 'failed',
152 error_msg = ?,
153 updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
154 where id = ?
155 `, errMsg, id)
156 return err
157}
158
159func ListGitRepoMigrationsForOwner(ctx context.Context, e Execer, owner string) (GitRepoMigrations, error) {
160 rows, err := e.QueryContext(ctx, `
161 select id, owner_did, source_kind, clone_url, name, knot,
162 description, session_id, status, coalesce(error_msg, ''),
163 updated_at
164 from gitrepo_migrations
165 where owner_did = ?
166 order by id desc
167 `, owner)
168 if err != nil {
169 return nil, err
170 }
171 defer rows.Close()
172
173 var out GitRepoMigrations
174 for rows.Next() {
175 var (
176 m GitRepoMigration
177 updatedAt string
178 )
179 if err := rows.Scan(
180 &m.ID, &m.OwnerDid, &m.SourceKind, &m.CloneUrl, &m.Name, &m.Knot,
181 &m.Description, &m.SessionID, &m.Status, &m.ErrorMsg,
182 &updatedAt,
183 ); err != nil {
184 return nil, err
185 }
186 if t, err := time.Parse(time.RFC3339, updatedAt); err == nil {
187 m.UpdatedAt = t
188 }
189 out = append(out, m)
190 }
191 return out, rows.Err()
192}
193
194func ListEnqueuedGitRepoNames(ctx context.Context, e Execer, owner string) (map[string]struct{}, error) {
195 rows, err := e.QueryContext(ctx, `
196 select name from gitrepo_migrations
197 where owner_did = ?
198 and status in ('pending', 'running', 'done')
199 `, owner)
200 if err != nil {
201 return nil, err
202 }
203 defer rows.Close()
204
205 out := map[string]struct{}{}
206 for rows.Next() {
207 var n string
208 if err := rows.Scan(&n); err != nil {
209 return nil, err
210 }
211 out[n] = struct{}{}
212 }
213 return out, rows.Err()
214}
215
216func ReapStaleRunningGitRepoMigrations(ctx context.Context, e Execer) error {
217 _, err := e.ExecContext(ctx, `
218 update gitrepo_migrations
219 set status = 'pending',
220 updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
221 where status = 'running'
222 `)
223 return err
224}