willdot.net
-2
cmd/main.go
-2
cmd/main.go
+80
-20
consumer.go
+80
-20
consumer.go
···
12
12
"github.com/bluesky-social/jetstream/pkg/client/schedulers/sequential"
13
13
"github.com/bluesky-social/jetstream/pkg/models"
14
14
"github.com/bugsnag/bugsnag-go"
15
-
"tangled.sh/tangled.sh/core/api/tangled"
15
+
"tangled.org/core/api/tangled"
16
16
)
17
17
18
18
type Issue struct {
···
29
29
RKey string `json:"rkey"`
30
30
Body string `json:"body"`
31
31
Issue string `json:"issue" `
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
+
DeleteIssue(did, rkey string) error
39
+
DeleteComment(did, rkey string) error
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
-
return h.handleCreateEvent(ctx, event)
106
-
// TODO: handle deletes too
107
+
return h.handleCreateUpdateEvent(ctx, event)
108
+
case models.CommitOperationDelete:
109
+
return h.handleDeleteEvent(ctx, event)
107
110
default:
108
111
return nil
109
112
}
110
113
}
111
114
112
-
func (h *Handler) handleCreateEvent(ctx context.Context, event *models.Event) error {
115
+
func (h *Handler) handleCreateUpdateEvent(ctx context.Context, event *models.Event) error {
113
116
switch event.Commit.Collection {
114
117
case tangled.RepoIssueNSID:
115
-
h.handleIssueEvent(ctx, event)
118
+
h.handleCreateUpdateIssueEvent(ctx, event)
116
119
case tangled.RepoIssueCommentNSID:
117
-
h.handleIssueCommentEvent(ctx, event)
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
-
func (h *Handler) handleIssueEvent(ctx context.Context, event *models.Event) {
129
+
func (h *Handler) handleDeleteEvent(ctx context.Context, event *models.Event) error {
130
+
switch event.Commit.Collection {
131
+
case tangled.RepoIssueNSID:
132
+
h.handleDeleteIssueEvent(ctx, event)
133
+
case tangled.RepoIssueCommentNSID:
134
+
h.handleDeleteIssueCommentEvent(ctx, event)
135
+
default:
136
+
slog.Info("create event was not for expected collection", "RKey", "did", event.Did, event.Commit.RKey, "collection", event.Commit.Collection)
137
+
return nil
138
+
}
139
+
140
+
return nil
141
+
}
142
+
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
-
func (h *Handler) handleIssueCommentEvent(ctx context.Context, event *models.Event) {
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
}
184
-
err = h.store.CreateComment(Comment{
185
-
AuthorDID: did,
186
-
RKey: rkey,
187
-
Body: comment.Body,
188
-
Issue: comment.Issue,
189
-
CreatedAt: createdAt.UnixMilli(),
190
-
//ReplyTo: comment, // TODO: there should be a ReplyTo field that can be used as well once the right type is imported
191
-
})
201
+
202
+
// if there is a replyTo present, don't store the comment because replies can't be replied to so
203
+
// the reply comment doesn't need to be stored
204
+
if comment.ReplyTo == nil || *comment.ReplyTo == "" {
205
+
err = h.store.CreateComment(Comment{
206
+
AuthorDID: did,
207
+
RKey: rkey,
208
+
Body: comment.Body,
209
+
Issue: comment.Issue,
210
+
CreatedAt: createdAt.UnixMilli(),
211
+
})
212
+
if err != nil {
213
+
bugsnag.Notify(err)
214
+
slog.Error("create comment", "error", err, "did", did, "rkey", rkey)
215
+
return
216
+
}
217
+
}
218
+
219
+
// TODO: now send a notification to either the issue creator or whoever the comment was a reply to
220
+
221
+
slog.Info("created comment ", "value", comment, "did", did, "rkey", rkey)
222
+
}
223
+
224
+
func (h *Handler) handleDeleteIssueEvent(ctx context.Context, event *models.Event) {
225
+
did := event.Did
226
+
rkey := event.Commit.RKey
227
+
228
+
err := h.store.DeleteIssue(did, rkey)
192
229
if err != nil {
193
230
bugsnag.Notify(err)
194
-
slog.Error("create comment", "error", err, "did", did, "rkey", rkey)
231
+
slog.Error("delete issue", "error", err, "did", did, "rkey", rkey)
195
232
return
196
233
}
197
234
198
-
// TODO: now send a notification to either the issue creator or whoever the comment was a reply to
235
+
// now attempt to delete any comments on that issue since they can't be replied to now.
236
+
// Note: if unsuccessful it doesn't matter because a deleted issue and its comments are
237
+
// not visible on the UI and so no one will be able to reply to them so this is just a
238
+
// cleanup operation
239
+
issueURI := fmt.Sprintf("at://%s/%s/%s", did, tangled.RepoIssueNSID, rkey)
240
+
err = h.store.DeleteCommentsForIssue(issueURI)
241
+
if err != nil {
242
+
bugsnag.Notify(err)
243
+
slog.Error("delete comments for issue", "error", err, "issue URI", issueURI)
244
+
}
199
245
200
-
slog.Info("created comment ", "value", comment, "did", did, "rkey", rkey)
246
+
slog.Info("deleted issue ", "did", did, "rkey", rkey)
247
+
}
248
+
249
+
func (h *Handler) handleDeleteIssueCommentEvent(ctx context.Context, event *models.Event) {
250
+
did := event.Did
251
+
rkey := event.Commit.RKey
252
+
253
+
err := h.store.DeleteComment(did, rkey)
254
+
if err != nil {
255
+
bugsnag.Notify(err)
256
+
slog.Error("delete comment", "error", err, "did", did, "rkey", rkey)
257
+
return
258
+
}
259
+
260
+
slog.Info("deleted comment ", "did", did, "rkey", rkey)
201
261
}
+31
-5
database.go
+31
-5
database.go
···
104
104
"rkey" TEXT,
105
105
"body" TEXT,
106
106
"issue" TEXT,
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
-
sql := `REPLACE INTO comments (authorDid, rkey, body, issue, replyTo, createdAt) VALUES (?, ?, ?, ?, ?, ?);`
139
-
_, err := d.db.Exec(sql, comment.AuthorDID, comment.RKey, comment.Body, comment.Issue, comment.ReplyTo, comment.CreatedAt)
137
+
sql := `REPLACE INTO comments (authorDid, rkey, body, issue, createdAt) VALUES (?, ?, ?, ?, ?);`
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
-
sql := "SELECT authorDid, rkey, body, issue, replyTo, createdAt FROM comments;"
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
-
if err := rows.Scan(&comment.AuthorDID, &comment.RKey, &comment.Body, &comment.Issue, &comment.ReplyTo, &comment.CreatedAt); err != nil {
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
+
185
+
func (d *Database) DeleteIssue(did, rkey string) error {
186
+
sql := "DELETE FROM issues WHERE authorDid = ? AND rkey = ?;"
187
+
_, err := d.db.Exec(sql, did, rkey)
188
+
if err != nil {
189
+
return fmt.Errorf("exec delete issue: %w", err)
190
+
}
191
+
return nil
192
+
}
193
+
194
+
func (d *Database) DeleteComment(did, rkey string) error {
195
+
sql := "DELETE FROM comments WHERE authorDid = ? AND rkey = ?;"
196
+
_, err := d.db.Exec(sql, did, rkey)
197
+
if err != nil {
198
+
return fmt.Errorf("exec delete issue: %w", err)
199
+
}
200
+
return nil
201
+
}
202
+
203
+
func (d *Database) DeleteCommentsForIssue(issueURI string) error {
204
+
sql := "DELETE FROM comments WHERE issue = ?;"
205
+
_, err := d.db.Exec(sql, issueURI)
206
+
if err != nil {
207
+
return fmt.Errorf("exec delete comments for issue")
208
+
}
209
+
return nil
210
+
}
+1
-1
go.mod
+1
-1
go.mod
···
8
8
github.com/bugsnag/bugsnag-go v2.6.2+incompatible
9
9
github.com/glebarez/go-sqlite v1.22.0
10
10
github.com/joho/godotenv v1.5.1
11
-
tangled.sh/tangled.sh/core v1.8.1-alpha.0.20250828210137-07b009bd6b98
11
+
tangled.org/core v1.9.0-alpha.0.20250924195920-24d79d05e4d2
12
12
)
13
13
14
14
require (
+4
-2
go.sum
+4
-2
go.sum
···
103
103
modernc.org/memory v1.7.2/go.mod h1:NO4NVCQy0N7ln+T9ngWqOQfi7ley4vpwvARR+Hjw95E=
104
104
modernc.org/sqlite v1.28.0 h1:Zx+LyDDmXczNnEQdvPuEfcFVA2ZPyaD7UCZDjef3BHQ=
105
105
modernc.org/sqlite v1.28.0/go.mod h1:Qxpazz0zH8Z1xCFyi5GSL3FzbtZ3fvbjmywNogldEW0=
106
-
tangled.sh/tangled.sh/core v1.8.1-alpha.0.20250828210137-07b009bd6b98 h1:WovrwwBufU89zoSaStoc6+qyUTEB/LxhUCM1MqGEUwU=
107
-
tangled.sh/tangled.sh/core v1.8.1-alpha.0.20250828210137-07b009bd6b98/go.mod h1:zXmPB9VMsPWpJ6Y51PWnzB1fL3w69P0IhR9rTXIfGPY=
106
+
tangled.org/core v1.9.0-alpha.0.20250924195920-24d79d05e4d2 h1:4bcQewZPzb7WfCuUPf4MPVWb04JiTbjbShcg5ONi9co=
107
+
tangled.org/core v1.9.0-alpha.0.20250924195920-24d79d05e4d2/go.mod h1:tYTB3RkgkeDAOFE0qX/9tQB80fdlDPR+vz4CdTMar3Y=
108
+
tangled.org/core v1.9.0-alpha.0.20250924200730-b2d8a54abc3d h1:DmdCyK+BZDYitJy6TdqTwvcci2EVYgDu2+LR853nyls=
109
+
tangled.org/core v1.9.0-alpha.0.20250924200730-b2d8a54abc3d/go.mod h1:tYTB3RkgkeDAOFE0qX/9tQB80fdlDPR+vz4CdTMar3Y=