-2

cmd/main.go

··· 86 86 if errors.Is(err, context.Canceled) { 87 87 return nil 88 88 } 89 89 - slog.Error("consume loop", "error", err) 90 90 - bugsnag.Notify(err) 91 89 return err 92 90 } 93 91 return nil

+68 -8

consumer.go

··· 29 29 RKey string `json:"rkey"` 30 30 Body string `json:"body"` 31 31 Issue string `json:"issue" ` 32 32 - ReplyTo string `json:"replyTo"` 33 32 CreatedAt int64 `json:"createdAt"` 34 33 } 35 34 36 35 type Store interface { 37 36 CreateIssue(issue Issue) error 38 37 CreateComment(comment Comment) error 38 38 + DeleteIssue(did, rkey string) error 39 39 + DeleteComment(did, rkey string) error 40 40 + DeleteCommentsForIssue(issueURI string) error 39 41 } 40 42 41 43 // JetstreamConsumer is responsible for consuming from a jetstream instance ··· 102 104 103 105 switch event.Commit.Operation { 104 106 case models.CommitOperationCreate, models.CommitOperationUpdate: 105 105 - return h.handleCreateEvent(ctx, event) 106 106 - // TODO: handle deletes too 107 107 + return h.handleCreateUpdateEvent(ctx, event) 108 108 + case models.CommitOperationDelete: 109 109 + return h.handleDeleteEvent(ctx, event) 107 110 default: 108 111 return nil 109 112 } 110 113 } 111 114 112 112 - func (h *Handler) handleCreateEvent(ctx context.Context, event *models.Event) error { 115 115 + func (h *Handler) handleCreateUpdateEvent(ctx context.Context, event *models.Event) error { 113 116 switch event.Commit.Collection { 114 117 case tangled.RepoIssueNSID: 115 115 - h.handleIssueEvent(ctx, event) 118 118 + h.handleCreateUpdateIssueEvent(ctx, event) 116 119 case tangled.RepoIssueCommentNSID: 117 117 - h.handleIssueCommentEvent(ctx, event) 120 120 + h.handleCreateUpdateIssueCommentEvent(ctx, event) 118 121 default: 119 122 slog.Info("create event was not for expected collection", "RKey", "did", event.Did, event.Commit.RKey, "collection", event.Commit.Collection) 120 123 return nil ··· 123 126 return nil 124 127 } 125 128 126 126 - func (h *Handler) handleIssueEvent(ctx context.Context, event *models.Event) { 129 129 + func (h *Handler) handleDeleteEvent(ctx context.Context, event *models.Event) error { 130 130 + switch event.Commit.Collection { 131 131 + case tangled.RepoIssueNSID: 132 132 + h.handleDeleteIssueEvent(ctx, event) 133 133 + case tangled.RepoIssueCommentNSID: 134 134 + h.handleDeleteIssueCommentEvent(ctx, event) 135 135 + default: 136 136 + slog.Info("create event was not for expected collection", "RKey", "did", event.Did, event.Commit.RKey, "collection", event.Commit.Collection) 137 137 + return nil 138 138 + } 139 139 + 140 140 + return nil 141 141 + } 142 142 + 143 143 + func (h *Handler) handleCreateUpdateIssueEvent(ctx context.Context, event *models.Event) { 127 144 var issue tangled.RepoIssue 128 145 129 146 err := json.Unmarshal(event.Commit.Record, &issue) ··· 162 179 slog.Info("created issue ", "value", issue, "did", did, "rkey", rkey) 163 180 } 164 181 165 165 - func (h *Handler) handleIssueCommentEvent(ctx context.Context, event *models.Event) { 182 182 + func (h *Handler) handleCreateUpdateIssueCommentEvent(ctx context.Context, event *models.Event) { 166 183 var comment tangled.RepoIssueComment 167 184 168 185 err := json.Unmarshal(event.Commit.Record, &comment) ··· 181 198 slog.Error("parsing createdAt time from comment", "error", err, "timestamp", comment.CreatedAt) 182 199 createdAt = time.Now().UTC() 183 200 } 201 201 + 202 202 + // TODO: if there is a reply to present, don't store the comment because replies can't be replied to so 203 203 + // the reply comment doesn't need to be stored 204 204 + 184 205 err = h.store.CreateComment(Comment{ 185 206 AuthorDID: did, 186 207 RKey: rkey, ··· 199 220 200 221 slog.Info("created comment ", "value", comment, "did", did, "rkey", rkey) 201 222 } 223 223 + 224 224 + func (h *Handler) handleDeleteIssueEvent(ctx context.Context, event *models.Event) { 225 225 + did := event.Did 226 226 + rkey := event.Commit.RKey 227 227 + 228 228 + err := h.store.DeleteIssue(did, rkey) 229 229 + if err != nil { 230 230 + bugsnag.Notify(err) 231 231 + slog.Error("delete issue", "error", err, "did", did, "rkey", rkey) 232 232 + return 233 233 + } 234 234 + 235 235 + // now attempt to delete any comments on that issue since they can't be replied to now. 236 236 + // Note: if unsuccessful it doesn't matter because a deleted issue and its comments are 237 237 + // not visible on the UI and so no one will be able to reply to them so this is just a 238 238 + // cleanup operation 239 239 + issueURI := fmt.Sprintf("at://%s/%s/%s", did, tangled.RepoIssueNSID, rkey) 240 240 + err = h.store.DeleteCommentsForIssue(issueURI) 241 241 + if err != nil { 242 242 + bugsnag.Notify(err) 243 243 + slog.Error("delete comments for issue", "error", err, "issue URI", issueURI) 244 244 + } 245 245 + 246 246 + slog.Info("deleted issue ", "did", did, "rkey", rkey) 247 247 + } 248 248 + 249 249 + func (h *Handler) handleDeleteIssueCommentEvent(ctx context.Context, event *models.Event) { 250 250 + did := event.Did 251 251 + rkey := event.Commit.RKey 252 252 + 253 253 + err := h.store.DeleteComment(did, rkey) 254 254 + if err != nil { 255 255 + bugsnag.Notify(err) 256 256 + slog.Error("delete comment", "error", err, "did", did, "rkey", rkey) 257 257 + return 258 258 + } 259 259 + 260 260 + slog.Info("deleted comment ", "did", did, "rkey", rkey) 261 261 + }

+31 -5

database.go

··· 104 104 "rkey" TEXT, 105 105 "body" TEXT, 106 106 "issue" TEXT, 107 107 - "replyTo" TEXT, 108 107 "createdAt" integer NOT NULL, 109 108 UNIQUE(authorDid,rkey) 110 109 );` ··· 135 134 136 135 // CreateComment will insert a comment into a database 137 136 func (d *Database) CreateComment(comment Comment) error { 138 138 - sql := `REPLACE INTO comments (authorDid, rkey, body, issue, replyTo, createdAt) VALUES (?, ?, ?, ?, ?, ?);` 139 139 - _, err := d.db.Exec(sql, comment.AuthorDID, comment.RKey, comment.Body, comment.Issue, comment.ReplyTo, comment.CreatedAt) 137 137 + sql := `REPLACE INTO comments (authorDid, rkey, body, issue, createdAt) VALUES (?, ?, ?, ?, ?);` 138 138 + _, err := d.db.Exec(sql, comment.AuthorDID, comment.RKey, comment.Body, comment.Issue, comment.CreatedAt) 140 139 if err != nil { 141 140 return fmt.Errorf("exec insert comment: %w", err) 142 141 } ··· 164 163 } 165 164 166 165 func (d *Database) GetComments() ([]Comment, error) { 167 167 - sql := "SELECT authorDid, rkey, body, issue, replyTo, createdAt FROM comments;" 166 166 + sql := "SELECT authorDid, rkey, body, issue, createdAt FROM comments;" 168 167 rows, err := d.db.Query(sql) 169 168 if err != nil { 170 169 return nil, fmt.Errorf("run query to get comments: %w", err) ··· 174 173 var results []Comment 175 174 for rows.Next() { 176 175 var comment Comment 177 177 - if err := rows.Scan(&comment.AuthorDID, &comment.RKey, &comment.Body, &comment.Issue, &comment.ReplyTo, &comment.CreatedAt); err != nil { 176 176 + if err := rows.Scan(&comment.AuthorDID, &comment.RKey, &comment.Body, &comment.Issue, &comment.CreatedAt); err != nil { 178 177 return nil, fmt.Errorf("scan row: %w", err) 179 178 } 180 179 ··· 182 181 } 183 182 return results, nil 184 183 } 184 184 + 185 185 + func (d *Database) DeleteIssue(did, rkey string) error { 186 186 + sql := "DELETE FROM issues WHERE authorDid = ? AND rkey = ?;" 187 187 + _, err := d.db.Exec(sql, did, rkey) 188 188 + if err != nil { 189 189 + return fmt.Errorf("exec delete issue: %w", err) 190 190 + } 191 191 + return nil 192 192 + } 193 193 + 194 194 + func (d *Database) DeleteComment(did, rkey string) error { 195 195 + sql := "DELETE FROM comments WHERE authorDid = ? AND rkey = ?;" 196 196 + _, err := d.db.Exec(sql, did, rkey) 197 197 + if err != nil { 198 198 + return fmt.Errorf("exec delete issue: %w", err) 199 199 + } 200 200 + return nil 201 201 + } 202 202 + 203 203 + func (d *Database) DeleteCommentsForIssue(issueURI string) error { 204 204 + sql := "DELETE FROM comments WHERE issue = ?;" 205 205 + _, err := d.db.Exec(sql, issueURI) 206 206 + if err != nil { 207 207 + return fmt.Errorf("exec delete comments for issue") 208 208 + } 209 209 + return nil 210 210 + }