ptr.pet
at protocol indexer with flexible filtering, xrpc queries, and a cursor-backed event stream, built on fjall
at-protocol
atproto
indexer
rust
fjall
[ingest] invalidate cached doc on identity event, update repo state
+19
-2
src/ingest/worker.rs
+19
-2
src/ingest/worker.rs
···
355
}
356
SubscribeReposMessage::Identity(identity) => {
357
debug!(did = %did, "processing buffered identity");
358
-
let handle = identity.handle.as_ref().map(|h| h.clone().into_static());
359
360
let evt = IdentityEvt {
361
did: did.clone().into_static(),
362
-
handle,
363
};
364
ctx.broadcast_events
365
.push(ops::make_identity_event(&ctx.state.db, evt));
···
355
}
356
SubscribeReposMessage::Identity(identity) => {
357
debug!(did = %did, "processing buffered identity");
358
+
359
+
if identity.handle.is_none() {
360
+
// we invalidate only if no handle is sent since its like a
361
+
// "invalidate your caches" message then basically
362
+
ctx.state.resolver.invalidate_sync(did);
363
+
let (_, handle) = ctx
364
+
.handle
365
+
.block_on(ctx.state.resolver.resolve_identity_info(did))?;
366
+
repo_state.handle = handle;
367
+
}
368
+
369
+
let handle = identity.handle.as_ref().map(|h| h.clone());
370
+
repo_state.handle = handle.or(repo_state.handle);
371
+
ctx.batch.insert(
372
+
&ctx.state.db.repos,
373
+
keys::repo_key(did),
374
+
crate::db::ser_repo_state(&repo_state)?,
375
+
);
376
377
let evt = IdentityEvt {
378
did: did.clone().into_static(),
379
+
handle: repo_state.handle.clone(),
380
};
381
ctx.broadcast_events
382
.push(ops::make_identity_event(&ctx.state.db, evt));