backend for xcvr appview
check if store collision, allowing for better deduplication of lexstream & channel init
+67
-29
+26
-14
server/internal/db/lexicon.go
+26
-14
server/internal/db/lexicon.go
···
127
127
return &p, nil
128
128
}
129
129
130
-
func (s *Store) StoreChannel(channel *types.Channel, ctx context.Context) error {
131
-
_, err := s.pool.Exec(ctx, `
130
+
func (s *Store) StoreChannel(channel *types.Channel, ctx context.Context) (wasNew bool, err error) {
131
+
commandTag, err := s.pool.Exec(ctx, `
132
132
INSERT INTO channels (
133
133
uri,
134
134
cid,
···
141
141
$1, $2, $3, $4, $5, $6, $7
142
142
) ON CONFLICT (uri) DO NOTHING
143
143
`, channel.URI, channel.CID, channel.DID, channel.Host, channel.Title, channel.Topic, channel.CreatedAt)
144
-
return err
144
+
if err != nil {
145
+
return
146
+
}
147
+
wasNew = commandTag.RowsAffected() > 0
148
+
return
145
149
}
146
150
147
151
func (s *Store) UpdateChannel(channel *types.Channel, ctx context.Context) error {
···
167
171
return err
168
172
}
169
173
170
-
func (s *Store) StoreMessage(message *types.Message, ctx context.Context) error {
171
-
_, err := s.pool.Exec(ctx, `
174
+
func (s *Store) StoreMessage(message *types.Message, ctx context.Context) (wasNew bool, err error) {
175
+
commandTag, err := s.pool.Exec(ctx, `
172
176
INSERT INTO messages (
173
177
uri,
174
178
cid,
···
182
186
$1, $2, $3, $4, $5, $6, $7, $8
183
187
) ON CONFLICT (uri) DO NOTHING
184
188
`, message.URI, message.CID, message.DID, message.SignetURI, message.Body, message.Nick, message.Color, message.PostedAt)
185
-
return err
189
+
if err != nil {
190
+
return
191
+
}
192
+
wasNew = commandTag.RowsAffected() > 0
193
+
return
186
194
}
187
195
188
196
func (s *Store) UpdateMessage(message *types.Message, ctx context.Context) error {
···
198
206
posted_at
199
207
) VALUES (
200
208
$1, $2, $3, $4, $5, $6, $7, $8
201
-
) ON CONFLICT (uri) DO NOTHING
209
+
)
202
210
`, message.URI, message.CID, message.DID, message.SignetURI, message.Body, message.Nick, message.Color, message.PostedAt)
203
211
return err
204
212
}
···
241
249
return channelURI, nil
242
250
}
243
251
244
-
func (s *Store) StoreSignet(signet *types.Signet, ctx context.Context) error {
245
-
_, err := s.pool.Exec(ctx, `
252
+
func (s *Store) StoreSignet(signet *types.Signet, ctx context.Context) (wasNew bool, err error) {
253
+
commandTag, err := s.pool.Exec(ctx, `
246
254
INSERT INTO signets (
247
255
uri,
248
256
issuer_did,
···
257
265
`, signet.URI, signet.IssuerDID, signet.AuthorHandle, signet.ChannelURI, signet.MessageID, signet.CID, signet.StartedAt)
258
266
if err != nil {
259
267
err = errors.New("SOMETHING BAD HAPPENED: " + err.Error())
268
+
return
260
269
}
261
-
return err
270
+
wasNew = commandTag.RowsAffected() > 0
271
+
return
262
272
}
263
273
264
274
func (s *Store) UpdateSignet(signet *types.Signet, ctx context.Context) error {
···
288
298
return err
289
299
}
290
300
291
-
func (s *Store) StoreImage(image *types.Image, ctx context.Context) error {
292
-
_, err := s.pool.Exec(ctx, `INSERT INTO images (
301
+
func (s *Store) StoreImage(image *types.Image, ctx context.Context) (wasNew bool, err error) {
302
+
commandTag, err := s.pool.Exec(ctx, `INSERT INTO images (
293
303
uri,
294
304
did,
295
305
signet_uri,
···
318
328
image.CID,
319
329
image.PostedAt)
320
330
if err != nil {
321
-
return errors.New("effor storing image: " + err.Error())
331
+
err = errors.New("effor storing image: " + err.Error())
332
+
return
322
333
}
323
-
return nil
334
+
wasNew = commandTag.RowsAffected() > 0
335
+
return
324
336
}
325
337
326
338
func (s *Store) UpdateImage(image *types.Image, ctx context.Context) error {
+2
-2
server/internal/handler/lrcHandlers.go
+2
-2
server/internal/handler/lrcHandlers.go
···
178
178
w.Header().Set("Content-Type", "application/json")
179
179
encoder := json.NewEncoder(w)
180
180
response := struct {
181
-
Blob interface{} `json:"blob"`
182
-
UUID string `json:"uuid"`
181
+
Blob any `json:"blob"`
182
+
UUID string `json:"uuid"`
183
183
}{
184
184
Blob: blob, UUID: uuid,
185
185
}
+11
-3
server/internal/recordmanager/channel.go
+11
-3
server/internal/recordmanager/channel.go
···
14
14
)
15
15
16
16
func (rm *RecordManager) AcceptChannel(c *types.Channel, ctx context.Context) error {
17
-
err := rm.storeChannel(c, ctx)
17
+
wasNew, err := rm.storeChannel(c, ctx)
18
18
if err != nil {
19
19
return errors.New("failed to store channel: " + err.Error())
20
+
}
21
+
if !wasNew {
22
+
return nil
20
23
}
21
24
err = rm.initChannel(c)
22
25
if err != nil {
···
72
75
err = errors.New("couldn't create channel: " + err.Error())
73
76
return
74
77
}
75
-
err = rm.storeChannel(channel, ctx)
78
+
wasNew, err := rm.storeChannel(channel, ctx)
76
79
if err != nil {
77
80
err = errors.New("couldn't store channel: " + err.Error())
81
+
return
82
+
}
83
+
if !wasNew {
84
+
did = channel.DID
85
+
uri = channel.URI
78
86
return
79
87
}
80
88
err = rm.initChannel(channel)
···
87
95
return
88
96
}
89
97
90
-
func (rm *RecordManager) storeChannel(c *types.Channel, ctx context.Context) error {
98
+
func (rm *RecordManager) storeChannel(c *types.Channel, ctx context.Context) (wasNew bool, err error) {
91
99
return rm.db.StoreChannel(c, ctx)
92
100
}
93
101
+4
-1
server/internal/recordmanager/media.go
+4
-1
server/internal/recordmanager/media.go
···
71
71
if err != nil {
72
72
return errors.New("coudlnt validate media record: " + err.Error())
73
73
}
74
-
err = rm.db.StoreImage(img, ctx)
74
+
wasNew, err := rm.db.StoreImage(img, ctx)
75
75
if err != nil {
76
76
return errors.New("beeped that up!: " + err.Error())
77
+
}
78
+
if !wasNew {
79
+
return nil
77
80
}
78
81
err = rm.forwardImage(img, ctx)
79
82
if err != nil {
+13
-4
server/internal/recordmanager/message.go
+13
-4
server/internal/recordmanager/message.go
···
16
16
)
17
17
18
18
func (rm *RecordManager) AcceptMessage(m *types.Message, ctx context.Context) error {
19
-
err := rm.storeMessage(m, ctx)
19
+
wasNew, err := rm.storeMessage(m, ctx)
20
20
if err != nil {
21
21
return errors.New("failed to store message: " + err.Error())
22
+
}
23
+
if !wasNew {
24
+
return nil
22
25
}
23
26
err = rm.forwardMessage(m, ctx)
24
27
if err != nil {
···
82
85
return errors.New("failed to create message: " + err.Error())
83
86
}
84
87
rm.log.Deprintln("store")
85
-
err = rm.storeMessage(m, ctx)
88
+
wasNew, err := rm.storeMessage(m, ctx)
86
89
if err != nil {
87
90
return errors.New("failed to store message: " + err.Error())
88
91
}
92
+
if !wasNew {
93
+
return nil
94
+
}
89
95
rm.log.Deprintln("forward")
90
96
err = rm.forwardMessage(m, ctx)
91
97
if err != nil {
···
107
113
if err != nil {
108
114
return errors.New("failed to create message: " + err.Error())
109
115
}
110
-
err = rm.storeMessage(m, ctx)
116
+
wasNew, err := rm.storeMessage(m, ctx)
111
117
if err != nil {
112
118
return errors.New("failed to store message: " + err.Error())
119
+
}
120
+
if !wasNew {
121
+
return nil
113
122
}
114
123
err = rm.forwardMessage(m, ctx)
115
124
if err != nil {
···
183
192
return rm.db.UpdateMessage(m, ctx)
184
193
}
185
194
186
-
func (rm *RecordManager) storeMessage(m *types.Message, ctx context.Context) error {
195
+
func (rm *RecordManager) storeMessage(m *types.Message, ctx context.Context) (wasNew bool, err error) {
187
196
return rm.db.StoreMessage(m, ctx)
188
197
}
189
198
+11
-5
server/internal/recordmanager/signet.go
+11
-5
server/internal/recordmanager/signet.go
···
20
20
if err != nil {
21
21
return errors.New("failed to create signet: " + err.Error())
22
22
}
23
-
err = rm.storeSignet(signet, ctx)
23
+
wasNew, err := rm.storeSignet(signet, ctx)
24
24
if err != nil {
25
25
return errors.New("failed to store signet: " + err.Error())
26
+
}
27
+
if !wasNew {
28
+
return nil
26
29
}
27
30
err = rm.forwardSignet(signet, uri)
28
31
if err != nil {
···
48
51
}
49
52
50
53
func (rm *RecordManager) AcceptSignet(s *types.Signet, ctx context.Context) error {
51
-
err := rm.storeSignet(s, ctx)
54
+
wasNew, err := rm.storeSignet(s, ctx)
52
55
if err != nil {
53
-
return errors.New("failed to store signet")
56
+
return errors.New("failed to store signet: " + err.Error())
57
+
}
58
+
if !wasNew {
59
+
return nil
54
60
}
55
-
return nil
61
+
return rm.forwardSignet(s, s.ChannelURI)
56
62
}
57
63
58
64
func (rm *RecordManager) AcceptSignetDelete(uri string, ctx context.Context) error {
···
104
110
return &sr, nil
105
111
}
106
112
107
-
func (rm *RecordManager) storeSignet(signet *types.Signet, ctx context.Context) error {
113
+
func (rm *RecordManager) storeSignet(signet *types.Signet, ctx context.Context) (bool, error) {
108
114
return rm.db.StoreSignet(signet, ctx)
109
115
}
110
116