183 unmodified lines
Expand all
ephemeral: state.ephemeral,
only_index_links: state.only_index_links,
count_ephemeral_records: matches!(origin, RecordEventOrigin::Backfill),
excluded,
has_redactions: None,
}
}
/// reads `did`'s repo row and keeps the pds it counts for, so a [`Self::put_repo`] of it
201 unmodified lines
Expand all
ephemeral: bool,
only_index_links: bool,
count_ephemeral_records: bool,
excluded: bool,
/// cached once per transaction step: almost no repo has operator redactions
has_redactions: Option<bool>,
}
/// a backfill step's record transaction on its owner shard, which only accepts a [`DiffStep`]
/// minted by `diff(new, StoredHeads)` where keys are strictly increasing and `Create` is
RecordWrite::Create {
collection,
rkey,
block,
} => self.0.put_record(&collection, &rkey, &block, DbAction::Create)?,
} => self.0.create_from_diff(&collection, &rkey, &block)?,
RecordWrite::Update {
collection,
rkey,
block,
} => self.0.put_record(&collection, &rkey, &block, DbAction::Update)?,
} => self.0.update_from_diff(&collection, &rkey, &block)?,
RecordWrite::Delete { collection, rkey } => {
self.0.delete_record(&collection, &rkey)?
self.0.delete_from_diff(&collection, &rkey)?
}
}
}
Ok(())
.as_ref()
.map(|old| crate::db::ipld_cid_from_record_value(old.as_ref()))
.transpose()?
.is_some_and(|old_cid| old_cid == cid);
let version_is_redacted = !self.only_index_links
&& self
.txn
.db
.indexer
.redactions
.contains_key(keys::redaction_key(&key, &cid))
.into_diagnostic()?;
let version_is_redacted = self.is_redacted(&key, &cid)?;
let store_cid_marker = self.only_index_links || version_is_redacted;
let head_is_cid_marker = head
.as_ref()
.is_some_and(|old| crate::db::is_cid_record_value(old));
self.count_record(collection, 1);
self.put_head(key, value);
}
}
self.index_and_emit(collection, rkey, action, cid, block, version_is_redacted)
}
fn count_record(&mut self, collection: &str, delta: i64) {
self.txn.counts.add_records(delta);
self.txn.add_record_count(self.did, collection, delta);
}
fn is_redacted(&mut self, key: &[u8], cid: &jacquard_common::types::cid::IpldCid) -> Result<bool> {
if self.only_index_links {
return Ok(false);
}
let has_any = match self.has_redactions {
Some(has) => has,
None => {
let prefix = keys::record_prefix_did(self.did);
let has = self
.txn
.db
.indexer
.redactions
.prefix(&prefix)
.next()
.map(|guard| guard.key())
.transpose()
.into_diagnostic()?
.is_some();
self.has_redactions = Some(has);
has
}
};
if !has_any {
return Ok(false);
}
self.txn
.db
.indexer
.redactions
.contains_key(keys::redaction_key(key, cid))
.into_diagnostic()
}
fn create_from_diff(
&mut self,
collection: &str,
rkey: &DbRkey,
block: &crate::car::CarBlock,
) -> Result<()> {
let block = block
.verify()
.map_err(|e| miette::miette!("record {}/{collection}/{rkey}: {e}", self.did))?;
let cid = *block.cid();
let block = block.bytes();
if self.ephemeral {
if self.count_ephemeral_records {
self.txn.counts.add_records(1);
}
return self.emitter.emit(
&mut self.txn.outbox,
EmitOp {
did: self.did,
collection,
rkey,
action: DbAction::Create,
cid: Some(cid),
block: Some(block),
},
);
}
let key = keys::record_key(self.did, collection, rkey);
let version_is_redacted = self.is_redacted(&key, &cid)?;
let cid_bytes = cid.to_bytes();
let value: &[u8] = if self.only_index_links || version_is_redacted {
&cid_bytes
} else {
block.as_ref()
};
self.count_record(collection, 1);
self.txn.batch.insert(&self.txn.db.indexer.heads, key, value);
self.index_and_emit(collection, rkey, DbAction::Create, cid, block, version_is_redacted)
}
fn update_from_diff(
&mut self,
collection: &str,
rkey: &DbRkey,
block: &crate::car::CarBlock,
) -> Result<()> {
let block = block
.verify()
.map_err(|e| miette::miette!("record {}/{collection}/{rkey}: {e}", self.did))?;
let cid = *block.cid();
let block = block.bytes();
if self.ephemeral {
return self.emitter.emit(
&mut self.txn.outbox,
EmitOp {
did: self.did,
collection,
rkey,
action: DbAction::Update,
cid: Some(cid),
block: Some(block),
},
);
}
let key = keys::record_key(self.did, collection, rkey);
let head = self.txn.db.indexer.head(&key).into_diagnostic()?;
let version_is_redacted = self.is_redacted(&key, &cid)?;
let cid_bytes = cid.to_bytes();
let value: &[u8] = if self.only_index_links || version_is_redacted {
&cid_bytes
} else {
block.as_ref()
};
match head {
Some(old) => {
if !self.only_index_links && !crate::db::is_cid_record_value(&old) {
self.txn.batch.insert(
&self.txn.db.indexer.history,
keys::history_key(&key, &self.death_rev(&key)),
old.as_ref(),
);
}
}
None => self.count_record(collection, 1),
}
self.txn.batch.insert(&self.txn.db.indexer.heads, key, value);
self.index_and_emit(collection, rkey, DbAction::Update, cid, block, version_is_redacted)
}
fn delete_from_diff(&mut self, collection: &str, rkey: &DbRkey) -> Result<()> {
if self.ephemeral {
if self.count_ephemeral_records {
self.txn.counts.add_records(-1);
}
return self.emitter.emit(
&mut self.txn.outbox,
EmitOp {
did: self.did,
collection,
rkey,
action: DbAction::Delete,
cid: None,
block: None,
},
);
}
let key = keys::record_key(self.did, collection, rkey);
if let Some(old) = self.txn.db.indexer.head(&key).into_diagnostic()? {
self.count_record(collection, -1);
if !self.only_index_links {
let history_value: &[u8] = if crate::db::is_cid_record_value(&old) {
&[]
} else {
old.as_ref()
};
self.txn.batch.insert(
&self.txn.db.indexer.history,
keys::history_key(&key, &self.death_rev(&key)),
history_value,
);
}
self.txn.batch.remove(&self.txn.db.indexer.heads, key);
}
crate::ops::backlink_ops::delete_record(
&mut self.txn.batch,
self.txn.db,
self.did,
collection,
&rkey.to_smolstr(),
)?;
self.emitter.emit(
&mut self.txn.outbox,
EmitOp {
did: self.did,
collection,
rkey,
action: DbAction::Delete,
cid: None,
block: None,
},
)
}
fn index_and_emit(
&mut self,
collection: &str,
rkey: &DbRkey,
action: DbAction,
cid: jacquard_common::types::cid::IpldCid,
block: &bytes::Bytes,
version_is_redacted: bool,
) -> Result<()> {
if version_is_redacted {
crate::ops::backlink_ops::delete_record(
&mut self.txn.batch,
self.txn.db,
&rkey.to_smolstr(),
block,
)?;
}
self.emitter.emit(
&mut self.txn.outbox,
EmitOp {
did: self.did,
},
)
}
fn count_record(&mut self, collection: &str, delta: i64) {
self.txn.counts.add_records(delta);
self.txn.add_record_count(self.did, collection, delta);
}
fn read_head(&self, key: &[u8]) -> Result<Option<fjall::Slice>> {
if let Some(staged) = self.txn.heads.get(key) {
return Ok(staged.clone());
}
1370 unmodified lines
Expand all