this repo has no description
deletes don't contain a reply, just an rkey so have to use that to do anything
+44
-22
+15
-15
consumer.go
+15
-15
consumer.go
···
103
103
// look for posts where I've "subsribed" so that we can add the parent URI to a list of replies to that parent to look for
104
104
if strings.Contains(post.Text, "/subscribe") && event.Did == "did:plc:dadhhalkfcq3gucaq25hjqon" {
105
105
slog.Info("a post that's subscribing to a parent. Adding to parents to look for", "parent URI", parentURI)
106
-
return h.addDidToSubscribedParent(parentURI, event.Did)
106
+
return h.addDidToSubscribedParent(parentURI, event.Did, event.Commit.RKey)
107
107
}
108
108
109
109
// see if the post is a reply to a post we are subscribed to
···
133
133
slog.Info("delete event received", "post", fmt.Sprintf("%+v", post))
134
134
}
135
135
136
-
// we only care about posts that have parents which are replies
137
-
if post.Reply == nil || post.Reply.Parent == nil || post.Reply.Parent.Uri == "" {
138
-
return nil
139
-
}
140
-
141
-
parentURI := post.Reply.Parent.Uri
142
-
143
-
// delete from subscriptions for the parentURI and the users DID
144
-
err := deleteFeedItemsForParentURIandUserDID(h.db, parentURI, event.Did)
136
+
parentURI, err := getSubscribingPostParentURI(h.db, event.Did, event.Commit.RKey)
145
137
if err != nil {
146
-
slog.Error("delete feed items for parentURI and user", "error", err, "parentURI", parentURI, "user DID", event.Did)
147
-
return fmt.Errorf("delete feed items for parentURI and user: %w", err)
138
+
slog.Error("get subscribing post parent URI", "error", err, "rkey", event.Commit.RKey, "user DID", event.Did)
139
+
return fmt.Errorf("get subscribing post parent URI: %w", err)
148
140
}
149
141
150
-
// delete from feeds for the parentURI and the users DID
142
+
// delete from feeds for the parentURI and the users DID first. This is so that if this fails, it can be tried again and the
143
+
// subscription will be still there
151
144
err = deleteSubscriptionForUser(h.db, event.Did, parentURI)
152
145
if err != nil {
153
146
slog.Error("delete subscription for user", "error", err, "parentURI", parentURI, "user DID", event.Did)
154
147
return fmt.Errorf("delete subscription and user: %w", err)
155
148
}
156
149
150
+
// delete from subscriptions for the parentURI and the users DID now that we have cleaned up the feeds
151
+
err = deleteFeedItemsForParentURIandUserDID(h.db, parentURI, event.Did)
152
+
if err != nil {
153
+
slog.Error("delete feed items for parentURI and user", "error", err, "parentURI", parentURI, "user DID", event.Did)
154
+
return fmt.Errorf("delete feed items for parentURI and user: %w", err)
155
+
}
156
+
157
157
return nil
158
158
}
159
159
160
-
func (h *handler) addDidToSubscribedParent(parentURI, userDid string) error {
161
-
err := addSubscriptionForParent(h.db, parentURI, userDid)
160
+
func (h *handler) addDidToSubscribedParent(parentURI, userDid, rkey string) error {
161
+
err := addSubscriptionForParent(h.db, parentURI, userDid, rkey)
162
162
if err != nil {
163
163
return fmt.Errorf("add subscription for parent: %w", err)
164
164
}
+29
-7
database.go
+29
-7
database.go
···
81
81
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
82
82
"parentURI" TEXT,
83
83
"userDID" TEXT,
84
-
UNIQUE(parentURI, userDID)
84
+
"subscriptionRkey" TEXT,
85
+
UNIQUE(parentURI, userDID, subscriptionRkey)
85
86
);`
86
87
87
88
slog.Info("Create subscriptions table...")
···
144
145
}
145
146
146
147
type subscription struct {
147
-
ID int
148
-
ParentURI string
149
-
UserDID string
148
+
ID int
149
+
ParentURI string
150
+
UserDID string
151
+
SubecriptionRkey string
150
152
}
151
153
152
154
func getSubscriptionsForParent(db *sql.DB, parentURI string) ([]string, error) {
···
169
171
return dids, nil
170
172
}
171
173
172
-
func addSubscriptionForParent(db *sql.DB, parentURI, userDid string) error {
173
-
sql := `INSERT INTO subscriptions (parentURI, userDID) VALUES (?, ?) ON CONFLICT(parentURI, userDID) DO NOTHING;`
174
-
_, err := db.Exec(sql, parentURI, userDid)
174
+
func addSubscriptionForParent(db *sql.DB, parentURI, userDid, subscriptionRkey string) error {
175
+
sql := `INSERT INTO subscriptions (parentURI, userDID, subscriptionRkey) VALUES (?, ?, ?) ON CONFLICT(parentURI, userDID) DO NOTHING;`
176
+
_, err := db.Exec(sql, parentURI, userDid, subscriptionRkey)
175
177
if err != nil {
176
178
return fmt.Errorf("exec insert subscrptions: %w", err)
177
179
}
178
180
return nil
181
+
}
182
+
183
+
func getSubscribingPostParentURI(db *sql.DB, userDID, rkey string) (string, error) {
184
+
sql := "SELECT parentURI FROM subscriptions WHERE (subscriptionRkey = ? AND userDID = ?);"
185
+
rows, err := db.Query(sql, rkey, userDID)
186
+
if err != nil {
187
+
return "", fmt.Errorf("run query to get subscribing post parent URI: %w", err)
188
+
}
189
+
defer rows.Close()
190
+
191
+
parentURI := ""
192
+
for rows.Next() {
193
+
var subscription subscription
194
+
if err := rows.Scan(&subscription.ID, &subscription.ParentURI, &subscription.UserDID); err != nil {
195
+
return "", fmt.Errorf("scan row: %w", err)
196
+
}
197
+
parentURI = subscription.ParentURI
198
+
break
199
+
}
200
+
return parentURI, nil
179
201
}
180
202
181
203
func deleteSubscriptionForUser(db *sql.DB, userDID, parentURI string) error {