Auto-indexing service and GraphQL API for AT Protocol Records
fix: transform BlobInput to AT Protocol format in create/update mutations
GraphQL BlobInput ({ref, mimeType, size}) must be transformed to
AT Protocol blob format ({$type: "blob", ref: {$link: ...}, ...})
before validation and sending to PDS.
+247
-15
+247
-15
server/src/mutation_resolvers.gleam
+247
-15
server/src/mutation_resolvers.gleam
···
8
8
import database/repositories/lexicons
9
9
import database/repositories/records
10
10
import dpop
11
+
import gleam/dict
11
12
import gleam/dynamic
12
13
import gleam/dynamic/decode
13
14
import gleam/erlang/process.{type Subject}
···
57
58
}
58
59
}
59
60
61
+
/// Get blob field paths from a lexicon for a given collection
62
+
/// Returns a list of paths, where each path is a list of field names
63
+
/// e.g., [["avatar"], ["banner"], ["nested", "image"]]
64
+
fn get_blob_paths(
65
+
collection: String,
66
+
lexicons: List(json.Json),
67
+
) -> List(List(String)) {
68
+
// Find the lexicon for this collection
69
+
let lexicon =
70
+
list.find(lexicons, fn(lex) {
71
+
case json.parse(json.to_string(lex), decode.at(["id"], decode.string)) {
72
+
Ok(id) -> id == collection
73
+
Error(_) -> False
74
+
}
75
+
})
76
+
77
+
case lexicon {
78
+
Ok(lex) -> {
79
+
// Get defs.main.record.properties
80
+
let properties_decoder =
81
+
decode.at(
82
+
["defs", "main", "record", "properties"],
83
+
decode.dict(decode.string, decode.dynamic),
84
+
)
85
+
case json.parse(json.to_string(lex), properties_decoder) {
86
+
Ok(properties) -> extract_blob_paths_from_properties(properties, [])
87
+
Error(_) -> []
88
+
}
89
+
}
90
+
Error(_) -> []
91
+
}
92
+
}
93
+
94
+
/// Recursively extract blob paths from lexicon properties
95
+
fn extract_blob_paths_from_properties(
96
+
properties: dict.Dict(String, dynamic.Dynamic),
97
+
current_path: List(String),
98
+
) -> List(List(String)) {
99
+
dict.fold(properties, [], fn(acc, field_name, field_def) {
100
+
let field_path = list.append(current_path, [field_name])
101
+
102
+
// Check field type
103
+
let type_result = decode.run(field_def, decode.at(["type"], decode.string))
104
+
105
+
case type_result {
106
+
Ok("blob") -> {
107
+
// Found a blob field
108
+
[field_path, ..acc]
109
+
}
110
+
Ok("object") -> {
111
+
// Recurse into nested object properties
112
+
let nested_props_result =
113
+
decode.run(
114
+
field_def,
115
+
decode.at(
116
+
["properties"],
117
+
decode.dict(decode.string, decode.dynamic),
118
+
),
119
+
)
120
+
case nested_props_result {
121
+
Ok(nested_props) -> {
122
+
let nested_paths =
123
+
extract_blob_paths_from_properties(nested_props, field_path)
124
+
list.append(nested_paths, acc)
125
+
}
126
+
Error(_) -> acc
127
+
}
128
+
}
129
+
Ok("array") -> {
130
+
// Check if array items are blobs or objects containing blobs
131
+
let items_type_result =
132
+
decode.run(field_def, decode.at(["items", "type"], decode.string))
133
+
case items_type_result {
134
+
Ok("blob") -> {
135
+
// Array of blobs - the path points to the array
136
+
[field_path, ..acc]
137
+
}
138
+
Ok("object") -> {
139
+
// Array of objects - recurse into item properties
140
+
let item_props_result =
141
+
decode.run(
142
+
field_def,
143
+
decode.at(
144
+
["items", "properties"],
145
+
decode.dict(decode.string, decode.dynamic),
146
+
),
147
+
)
148
+
case item_props_result {
149
+
Ok(item_props) -> {
150
+
let nested_paths =
151
+
extract_blob_paths_from_properties(item_props, field_path)
152
+
list.append(nested_paths, acc)
153
+
}
154
+
Error(_) -> acc
155
+
}
156
+
}
157
+
_ -> acc
158
+
}
159
+
}
160
+
_ -> acc
161
+
}
162
+
})
163
+
}
164
+
165
+
/// Transform blob inputs in a value from GraphQL format to AT Protocol format
166
+
/// GraphQL: { ref: "bafyrei...", mimeType: "image/jpeg", size: 12345 }
167
+
/// AT Proto: { "$type": "blob", ref: { "$link": "bafyrei..." }, mimeType: "image/jpeg", size: 12345 }
168
+
fn transform_blob_inputs(
169
+
input: value.Value,
170
+
blob_paths: List(List(String)),
171
+
) -> value.Value {
172
+
transform_value_at_paths(input, blob_paths, [])
173
+
}
174
+
175
+
/// Recursively transform values at blob paths
176
+
fn transform_value_at_paths(
177
+
val: value.Value,
178
+
blob_paths: List(List(String)),
179
+
current_path: List(String),
180
+
) -> value.Value {
181
+
case val {
182
+
value.Object(fields) -> {
183
+
// Check if current path matches any blob path exactly
184
+
let is_blob_path =
185
+
list.any(blob_paths, fn(path) {
186
+
path == current_path && current_path != []
187
+
})
188
+
189
+
case is_blob_path {
190
+
True -> {
191
+
// Transform this object from BlobInput to AT Protocol format
192
+
transform_blob_object(fields)
193
+
}
194
+
False -> {
195
+
// Recurse into object fields
196
+
value.Object(
197
+
list.map(fields, fn(field) {
198
+
let #(key, field_val) = field
199
+
let new_path = list.append(current_path, [key])
200
+
#(key, transform_value_at_paths(field_val, blob_paths, new_path))
201
+
}),
202
+
)
203
+
}
204
+
}
205
+
}
206
+
value.List(items) -> {
207
+
// Check if current path is a blob array path
208
+
let is_blob_array_path =
209
+
list.any(blob_paths, fn(path) {
210
+
path == current_path && current_path != []
211
+
})
212
+
213
+
case is_blob_array_path {
214
+
True -> {
215
+
// Transform each item as a blob
216
+
value.List(
217
+
list.map(items, fn(item) {
218
+
case item {
219
+
value.Object(item_fields) -> transform_blob_object(item_fields)
220
+
_ -> item
221
+
}
222
+
}),
223
+
)
224
+
}
225
+
False -> {
226
+
// Check if any blob path starts with current path (for arrays of objects)
227
+
let paths_through_here =
228
+
list.filter(blob_paths, fn(path) {
229
+
list.length(path) > list.length(current_path)
230
+
&& list.take(path, list.length(current_path)) == current_path
231
+
})
232
+
233
+
case list.is_empty(paths_through_here) {
234
+
True -> val
235
+
False -> {
236
+
// Recurse into array items with the same path
237
+
value.List(
238
+
list.map(items, fn(item) {
239
+
transform_value_at_paths(item, blob_paths, current_path)
240
+
}),
241
+
)
242
+
}
243
+
}
244
+
}
245
+
}
246
+
}
247
+
_ -> val
248
+
}
249
+
}
250
+
251
+
/// Transform a BlobInput object to AT Protocol blob format
252
+
fn transform_blob_object(fields: List(#(String, value.Value))) -> value.Value {
253
+
// Extract ref, mimeType, size from fields
254
+
let ref = case list.key_find(fields, "ref") {
255
+
Ok(value.String(r)) -> r
256
+
_ -> ""
257
+
}
258
+
let mime_type = case list.key_find(fields, "mimeType") {
259
+
Ok(value.String(m)) -> m
260
+
_ -> ""
261
+
}
262
+
let size = case list.key_find(fields, "size") {
263
+
Ok(value.Int(s)) -> s
264
+
_ -> 0
265
+
}
266
+
267
+
// Only transform if it looks like a valid BlobInput
268
+
case ref != "" && mime_type != "" {
269
+
True ->
270
+
value.Object([
271
+
#("$type", value.String("blob")),
272
+
#("ref", value.Object([#("$link", value.String(ref))])),
273
+
#("mimeType", value.String(mime_type)),
274
+
#("size", value.Int(size)),
275
+
])
276
+
False ->
277
+
// Not a valid BlobInput, return as-is
278
+
value.Object(fields)
279
+
}
280
+
}
281
+
60
282
/// Create a resolver factory for create mutations
61
283
pub fn create_resolver_factory(
62
284
collection: String,
···
151
373
}),
152
374
)
153
375
154
-
// Step 5: Convert input to JSON for validation and AT Protocol
155
-
let record_json_value = graphql_value_to_json_value(input)
156
-
let record_json_string = json.to_string(record_json_value)
157
-
158
-
// Step 6: Validate against lexicon (fetch all lexicons to resolve refs)
376
+
// Step 5: Fetch lexicons for validation and blob path extraction
159
377
use all_lexicon_records <- result.try(
160
378
lexicons.get_all(ctx.db)
161
379
|> result.map_error(fn(_) { "Failed to fetch lexicons" }),
···
172
390
}),
173
391
)
174
392
393
+
// Step 6: Transform blob inputs from GraphQL format to AT Protocol format
394
+
let blob_paths = get_blob_paths(collection, all_lex_jsons)
395
+
let transformed_input = transform_blob_inputs(input, blob_paths)
396
+
397
+
// Convert transformed input to JSON for validation and AT Protocol
398
+
let record_json_value = graphql_value_to_json_value(transformed_input)
399
+
let record_json_string = json.to_string(record_json_value)
400
+
401
+
// Step 7: Validate against lexicon
175
402
use _ <- result.try(
176
403
honk.validate_record(all_lex_jsons, collection, record_json_value)
177
404
|> result.map_error(fn(err) {
···
180
407
)
181
408
182
409
{
183
-
// Step 7: Call createRecord via AT Protocol
410
+
// Step 8: Call createRecord via AT Protocol
184
411
// Omit rkey field when not provided to let PDS auto-generate TID
185
412
let create_body =
186
413
case rkey {
···
189
416
#("repo", json.string(user_info.did)),
190
417
#("collection", json.string(collection)),
191
418
#("rkey", json.string(r)),
192
-
#("record", graphql_value_to_json_value(input)),
419
+
#("record", record_json_value),
193
420
])
194
421
option.None ->
195
422
json.object([
196
423
#("repo", json.string(user_info.did)),
197
424
#("collection", json.string(collection)),
198
-
#("record", graphql_value_to_json_value(input)),
425
+
#("record", record_json_value),
199
426
])
200
427
}
201
428
|> json.to_string
···
361
588
}),
362
589
)
363
590
364
-
// Step 5: Convert input to JSON for validation and AT Protocol
365
-
let record_json_value = graphql_value_to_json_value(input)
366
-
let record_json_string = json.to_string(record_json_value)
367
-
368
-
// Step 6: Validate against lexicon (fetch all lexicons to resolve refs)
591
+
// Step 5: Fetch lexicons for validation and blob path extraction
369
592
use all_lexicon_records <- result.try(
370
593
lexicons.get_all(ctx.db)
371
594
|> result.map_error(fn(_) { "Failed to fetch lexicons" }),
···
382
605
}),
383
606
)
384
607
608
+
// Step 6: Transform blob inputs from GraphQL format to AT Protocol format
609
+
let blob_paths = get_blob_paths(collection, all_lex_jsons)
610
+
let transformed_input = transform_blob_inputs(input, blob_paths)
611
+
612
+
// Convert transformed input to JSON for validation and AT Protocol
613
+
let record_json_value = graphql_value_to_json_value(transformed_input)
614
+
let record_json_string = json.to_string(record_json_value)
615
+
616
+
// Step 7: Validate against lexicon
385
617
use _ <- result.try(
386
618
honk.validate_record(all_lex_jsons, collection, record_json_value)
387
619
|> result.map_error(fn(err) {
···
390
622
)
391
623
392
624
{
393
-
// Step 7: Call putRecord via AT Protocol
625
+
// Step 8: Call putRecord via AT Protocol
394
626
let update_body =
395
627
json.object([
396
628
#("repo", json.string(user_info.did)),
397
629
#("collection", json.string(collection)),
398
630
#("rkey", json.string(rkey)),
399
-
#("record", graphql_value_to_json_value(input)),
631
+
#("record", record_json_value),
400
632
])
401
633
|> json.to_string
402
634