at master 549 lines 12 kB view raw
1package db 2 3import ( 4 "database/sql" 5 "fmt" 6 "maps" 7 "slices" 8 "sort" 9 "strings" 10 "time" 11 12 "github.com/bluesky-social/indigo/atproto/syntax" 13 "tangled.org/core/api/tangled" 14 "tangled.org/core/appview/models" 15 "tangled.org/core/appview/pagination" 16 "tangled.org/core/orm" 17) 18 19func PutIssue(tx *sql.Tx, issue *models.Issue) error { 20 // ensure sequence exists 21 _, err := tx.Exec(` 22 insert or ignore into repo_issue_seqs (repo_at, next_issue_id) 23 values (?, 1) 24 `, issue.RepoAt) 25 if err != nil { 26 return err 27 } 28 29 issues, err := GetIssues( 30 tx, 31 orm.FilterEq("did", issue.Did), 32 orm.FilterEq("rkey", issue.Rkey), 33 ) 34 switch { 35 case err != nil: 36 return err 37 case len(issues) == 0: 38 return createNewIssue(tx, issue) 39 case len(issues) != 1: // should be unreachable 40 return fmt.Errorf("invalid number of issues returned: %d", len(issues)) 41 default: 42 // if content is identical, do not edit 43 existingIssue := issues[0] 44 if existingIssue.Title == issue.Title && existingIssue.Body == issue.Body { 45 return nil 46 } 47 48 issue.Id = existingIssue.Id 49 issue.IssueId = existingIssue.IssueId 50 return updateIssue(tx, issue) 51 } 52} 53 54func createNewIssue(tx *sql.Tx, issue *models.Issue) error { 55 // get next issue_id 56 var newIssueId int 57 err := tx.QueryRow(` 58 update repo_issue_seqs 59 set next_issue_id = next_issue_id + 1 60 where repo_at = ? 61 returning next_issue_id - 1 62 `, issue.RepoAt).Scan(&newIssueId) 63 if err != nil { 64 return err 65 } 66 67 // insert new issue 68 row := tx.QueryRow(` 69 insert into issues (repo_at, did, rkey, issue_id, title, body) 70 values (?, ?, ?, ?, ?, ?) 71 returning rowid, issue_id 72 `, issue.RepoAt, issue.Did, issue.Rkey, newIssueId, issue.Title, issue.Body) 73 74 err = row.Scan(&issue.Id, &issue.IssueId) 75 if err != nil { 76 return fmt.Errorf("scan row: %w", err) 77 } 78 79 if err := putReferences(tx, issue.AtUri(), issue.References); err != nil { 80 return fmt.Errorf("put reference_links: %w", err) 81 } 82 return nil 83} 84 85func updateIssue(tx *sql.Tx, issue *models.Issue) error { 86 // update existing issue 87 _, err := tx.Exec(` 88 update issues 89 set title = ?, body = ?, edited = ? 90 where did = ? and rkey = ? 91 `, issue.Title, issue.Body, time.Now().Format(time.RFC3339), issue.Did, issue.Rkey) 92 if err != nil { 93 return err 94 } 95 96 if err := putReferences(tx, issue.AtUri(), issue.References); err != nil { 97 return fmt.Errorf("put reference_links: %w", err) 98 } 99 return nil 100} 101 102func GetIssuesPaginated(e Execer, page pagination.Page, filters ...orm.Filter) ([]models.Issue, error) { 103 issueMap := make(map[string]*models.Issue) // at-uri -> issue 104 105 var conditions []string 106 var args []any 107 108 for _, filter := range filters { 109 conditions = append(conditions, filter.Condition()) 110 args = append(args, filter.Arg()...) 111 } 112 113 whereClause := "" 114 if conditions != nil { 115 whereClause = " where " + strings.Join(conditions, " and ") 116 } 117 118 pLower := orm.FilterGte("row_num", page.Offset+1) 119 pUpper := orm.FilterLte("row_num", page.Offset+page.Limit) 120 121 pageClause := "" 122 if page.Limit > 0 { 123 args = append(args, pLower.Arg()...) 124 args = append(args, pUpper.Arg()...) 125 pageClause = " where " + pLower.Condition() + " and " + pUpper.Condition() 126 } 127 128 query := fmt.Sprintf( 129 ` 130 select * from ( 131 select 132 id, 133 did, 134 rkey, 135 repo_at, 136 issue_id, 137 title, 138 body, 139 open, 140 created, 141 edited, 142 deleted, 143 row_number() over (order by created desc) as row_num 144 from 145 issues 146 %s 147 ) ranked_issues 148 %s 149 `, 150 whereClause, 151 pageClause, 152 ) 153 154 rows, err := e.Query(query, args...) 155 if err != nil { 156 return nil, fmt.Errorf("failed to query issues table: %w", err) 157 } 158 defer rows.Close() 159 160 for rows.Next() { 161 var issue models.Issue 162 var createdAt string 163 var editedAt, deletedAt sql.Null[string] 164 var rowNum int64 165 err := rows.Scan( 166 &issue.Id, 167 &issue.Did, 168 &issue.Rkey, 169 &issue.RepoAt, 170 &issue.IssueId, 171 &issue.Title, 172 &issue.Body, 173 &issue.Open, 174 &createdAt, 175 &editedAt, 176 &deletedAt, 177 &rowNum, 178 ) 179 if err != nil { 180 return nil, fmt.Errorf("failed to scan issue: %w", err) 181 } 182 183 if t, err := time.Parse(time.RFC3339, createdAt); err == nil { 184 issue.Created = t 185 } 186 187 if editedAt.Valid { 188 if t, err := time.Parse(time.RFC3339, editedAt.V); err == nil { 189 issue.Edited = &t 190 } 191 } 192 193 if deletedAt.Valid { 194 if t, err := time.Parse(time.RFC3339, deletedAt.V); err == nil { 195 issue.Deleted = &t 196 } 197 } 198 199 atUri := issue.AtUri().String() 200 issueMap[atUri] = &issue 201 } 202 203 // collect reverse repos 204 repoAts := make([]string, 0, len(issueMap)) // or just []string{} 205 for _, issue := range issueMap { 206 repoAts = append(repoAts, string(issue.RepoAt)) 207 } 208 209 repos, err := GetRepos(e, 0, orm.FilterIn("at_uri", repoAts)) 210 if err != nil { 211 return nil, fmt.Errorf("failed to build repo mappings: %w", err) 212 } 213 214 repoMap := make(map[string]*models.Repo) 215 for i := range repos { 216 repoMap[string(repos[i].RepoAt())] = &repos[i] 217 } 218 219 for issueAt, i := range issueMap { 220 if r, ok := repoMap[string(i.RepoAt)]; ok { 221 i.Repo = r 222 } else { 223 // do not show up the issue if the repo is deleted 224 // TODO: foreign key where? 225 delete(issueMap, issueAt) 226 } 227 } 228 229 // collect comments 230 issueAts := slices.Collect(maps.Keys(issueMap)) 231 232 comments, err := GetIssueComments(e, orm.FilterIn("issue_at", issueAts)) 233 if err != nil { 234 return nil, fmt.Errorf("failed to query comments: %w", err) 235 } 236 for i := range comments { 237 issueAt := comments[i].IssueAt 238 if issue, ok := issueMap[issueAt]; ok { 239 issue.Comments = append(issue.Comments, comments[i]) 240 } 241 } 242 243 // collect allLabels for each issue 244 allLabels, err := GetLabels(e, orm.FilterIn("subject", issueAts)) 245 if err != nil { 246 return nil, fmt.Errorf("failed to query labels: %w", err) 247 } 248 for issueAt, labels := range allLabels { 249 if issue, ok := issueMap[issueAt.String()]; ok { 250 issue.Labels = labels 251 } 252 } 253 254 // collect references for each issue 255 allReferencs, err := GetReferencesAll(e, orm.FilterIn("from_at", issueAts)) 256 if err != nil { 257 return nil, fmt.Errorf("failed to query reference_links: %w", err) 258 } 259 for issueAt, references := range allReferencs { 260 if issue, ok := issueMap[issueAt.String()]; ok { 261 issue.References = references 262 } 263 } 264 265 var issues []models.Issue 266 for _, i := range issueMap { 267 issues = append(issues, *i) 268 } 269 270 sort.Slice(issues, func(i, j int) bool { 271 return issues[i].Created.After(issues[j].Created) 272 }) 273 274 return issues, nil 275} 276 277func GetIssue(e Execer, repoAt syntax.ATURI, issueId int) (*models.Issue, error) { 278 issues, err := GetIssuesPaginated( 279 e, 280 pagination.Page{}, 281 orm.FilterEq("repo_at", repoAt), 282 orm.FilterEq("issue_id", issueId), 283 ) 284 if err != nil { 285 return nil, err 286 } 287 if len(issues) != 1 { 288 return nil, sql.ErrNoRows 289 } 290 291 return &issues[0], nil 292} 293 294func GetIssues(e Execer, filters ...orm.Filter) ([]models.Issue, error) { 295 return GetIssuesPaginated(e, pagination.Page{}, filters...) 296} 297 298func AddIssueComment(tx *sql.Tx, c models.IssueComment) (int64, error) { 299 result, err := tx.Exec( 300 `insert into issue_comments ( 301 did, 302 rkey, 303 issue_at, 304 body, 305 reply_to, 306 created, 307 edited 308 ) 309 values (?, ?, ?, ?, ?, ?, null) 310 on conflict(did, rkey) do update set 311 issue_at = excluded.issue_at, 312 body = excluded.body, 313 edited = case 314 when 315 issue_comments.issue_at != excluded.issue_at 316 or issue_comments.body != excluded.body 317 or issue_comments.reply_to != excluded.reply_to 318 then ? 319 else issue_comments.edited 320 end`, 321 c.Did, 322 c.Rkey, 323 c.IssueAt, 324 c.Body, 325 c.ReplyTo, 326 c.Created.Format(time.RFC3339), 327 time.Now().Format(time.RFC3339), 328 ) 329 if err != nil { 330 return 0, err 331 } 332 333 id, err := result.LastInsertId() 334 if err != nil { 335 return 0, err 336 } 337 338 if err := putReferences(tx, c.AtUri(), c.References); err != nil { 339 return 0, fmt.Errorf("put reference_links: %w", err) 340 } 341 342 return id, nil 343} 344 345func DeleteIssueComments(e Execer, filters ...orm.Filter) error { 346 var conditions []string 347 var args []any 348 for _, filter := range filters { 349 conditions = append(conditions, filter.Condition()) 350 args = append(args, filter.Arg()...) 351 } 352 353 whereClause := "" 354 if conditions != nil { 355 whereClause = " where " + strings.Join(conditions, " and ") 356 } 357 358 query := fmt.Sprintf(`update issue_comments set body = "", deleted = strftime('%%Y-%%m-%%dT%%H:%%M:%%SZ', 'now') %s`, whereClause) 359 360 _, err := e.Exec(query, args...) 361 return err 362} 363 364func GetIssueComments(e Execer, filters ...orm.Filter) ([]models.IssueComment, error) { 365 commentMap := make(map[string]*models.IssueComment) 366 367 var conditions []string 368 var args []any 369 for _, filter := range filters { 370 conditions = append(conditions, filter.Condition()) 371 args = append(args, filter.Arg()...) 372 } 373 374 whereClause := "" 375 if conditions != nil { 376 whereClause = " where " + strings.Join(conditions, " and ") 377 } 378 379 query := fmt.Sprintf(` 380 select 381 id, 382 did, 383 rkey, 384 issue_at, 385 reply_to, 386 body, 387 created, 388 edited, 389 deleted 390 from 391 issue_comments 392 %s 393 `, whereClause) 394 395 rows, err := e.Query(query, args...) 396 if err != nil { 397 return nil, err 398 } 399 defer rows.Close() 400 401 for rows.Next() { 402 var comment models.IssueComment 403 var created string 404 var rkey, edited, deleted, replyTo sql.Null[string] 405 err := rows.Scan( 406 &comment.Id, 407 &comment.Did, 408 &rkey, 409 &comment.IssueAt, 410 &replyTo, 411 &comment.Body, 412 &created, 413 &edited, 414 &deleted, 415 ) 416 if err != nil { 417 return nil, err 418 } 419 420 // this is a remnant from old times, newer comments always have rkey 421 if rkey.Valid { 422 comment.Rkey = rkey.V 423 } 424 425 if t, err := time.Parse(time.RFC3339, created); err == nil { 426 comment.Created = t 427 } 428 429 if edited.Valid { 430 if t, err := time.Parse(time.RFC3339, edited.V); err == nil { 431 comment.Edited = &t 432 } 433 } 434 435 if deleted.Valid { 436 if t, err := time.Parse(time.RFC3339, deleted.V); err == nil { 437 comment.Deleted = &t 438 } 439 } 440 441 if replyTo.Valid { 442 comment.ReplyTo = &replyTo.V 443 } 444 445 atUri := comment.AtUri().String() 446 commentMap[atUri] = &comment 447 } 448 449 if err = rows.Err(); err != nil { 450 return nil, err 451 } 452 453 // collect references for each comments 454 commentAts := slices.Collect(maps.Keys(commentMap)) 455 allReferencs, err := GetReferencesAll(e, orm.FilterIn("from_at", commentAts)) 456 if err != nil { 457 return nil, fmt.Errorf("failed to query reference_links: %w", err) 458 } 459 for commentAt, references := range allReferencs { 460 if comment, ok := commentMap[commentAt.String()]; ok { 461 comment.References = references 462 } 463 } 464 465 var comments []models.IssueComment 466 for _, c := range commentMap { 467 comments = append(comments, *c) 468 } 469 470 sort.Slice(comments, func(i, j int) bool { 471 return comments[i].Created.After(comments[j].Created) 472 }) 473 474 return comments, nil 475} 476 477func DeleteIssues(tx *sql.Tx, did, rkey string) error { 478 _, err := tx.Exec( 479 `delete from issues 480 where did = ? and rkey = ?`, 481 did, 482 rkey, 483 ) 484 if err != nil { 485 return fmt.Errorf("delete issue: %w", err) 486 } 487 488 uri := syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", did, tangled.RepoIssueNSID, rkey)) 489 err = deleteReferences(tx, uri) 490 if err != nil { 491 return fmt.Errorf("delete reference_links: %w", err) 492 } 493 494 return nil 495} 496 497func CloseIssues(e Execer, filters ...orm.Filter) error { 498 var conditions []string 499 var args []any 500 for _, filter := range filters { 501 conditions = append(conditions, filter.Condition()) 502 args = append(args, filter.Arg()...) 503 } 504 505 whereClause := "" 506 if conditions != nil { 507 whereClause = " where " + strings.Join(conditions, " and ") 508 } 509 510 query := fmt.Sprintf(`update issues set open = 0 %s`, whereClause) 511 _, err := e.Exec(query, args...) 512 return err 513} 514 515func ReopenIssues(e Execer, filters ...orm.Filter) error { 516 var conditions []string 517 var args []any 518 for _, filter := range filters { 519 conditions = append(conditions, filter.Condition()) 520 args = append(args, filter.Arg()...) 521 } 522 523 whereClause := "" 524 if conditions != nil { 525 whereClause = " where " + strings.Join(conditions, " and ") 526 } 527 528 query := fmt.Sprintf(`update issues set open = 1 %s`, whereClause) 529 _, err := e.Exec(query, args...) 530 return err 531} 532 533func GetIssueCount(e Execer, repoAt syntax.ATURI) (models.IssueCount, error) { 534 row := e.QueryRow(` 535 select 536 count(case when open = 1 then 1 end) as open_count, 537 count(case when open = 0 then 1 end) as closed_count 538 from issues 539 where repo_at = ?`, 540 repoAt, 541 ) 542 543 var count models.IssueCount 544 if err := row.Scan(&count.Open, &count.Closed); err != nil { 545 return models.IssueCount{}, err 546 } 547 548 return count, nil 549}