use miette::Result;
use tracing::{debug, warn};
use url::Url;
use crate::types::{RepoState, RepoStatus};
use crate::ingest::stream::{Account, AccountStatus, Commit, Identity, Sync};
use crate::ingest::validation::ValidatedCommit;
use super::context::{Checked, Fact, RefreshOrigin};
use super::lookup::LookupKind;
use super::{RelayWorker, WorkerContext};
/// how a handler left its message: handled, or handed back to wait for a lookup
pub(crate) enum Handled<T> {
Done,
Wait(LookupKind, T),
}
impl RelayWorker {
pub(crate) fn handle_commit(
ctx: &mut WorkerContext,
mut repo_state: RepoState<'static>,
firehose: &Url,
commit: Box<Commit<'static>>,
) -> Result<Handled<Box<Commit<'static>>>> {
if !repo_state.active {
return Ok(Handled::Done);
}
repo_state.advance_message_time(commit.time.0.timestamp_millis());
let validated = match ctx.validate_commit(&mut repo_state, firehose, &commit)? {
Checked::Done(Some(validated)) => validated,
Checked::Done(None) => return Ok(Handled::Done),
Checked::Wait(kind) => return Ok(Handled::Wait(kind, commit)),
};
let ValidatedCommit {
chain_break,
commit_obj,
parsed_blocks,
..
} = validated;
if chain_break.is_broken() {
// chain breaks are not grounds for blocking when acting as a relay
debug!(broken = ?chain_break, "chain break, forwarding anyway");
}
ctx.sink.commit(
ctx.state,
&mut ctx.batch,
firehose,
*commit,
chain_break.is_broken(),
parsed_blocks,
)?;
repo_state.root = Some(commit_obj.into());
repo_state.touch();
ctx.stage_repo_state(repo_state);
Ok(Handled::Done)
}
pub(crate) fn handle_sync(
ctx: &mut WorkerContext,
mut repo_state: RepoState<'static>,
firehose: &Url,
sync: Box<Sync<'static>>,
) -> Result<Handled<Box<Sync<'static>>>> {
if !repo_state.active {
return Ok(Handled::Done);
}
// replaying an old #sync would make the indexer refetch every time
if let Some(root) = &repo_state.root
&& sync.rev.as_str() <= root.rev.to_tid().as_str()
{
debug!(rev = %sync.rev, "sync at or below the tracked rev, ignoring");
return Ok(Handled::Done);
}
repo_state.advance_message_time(sync.time.0.timestamp_millis());
let validated = match ctx.validate_sync(&mut repo_state, firehose, &sync)? {
Checked::Done(Some(validated)) => validated,
Checked::Done(None) => return Ok(Handled::Done),
Checked::Wait(kind) => return Ok(Handled::Wait(kind, sync)),
};
ctx.sink.sync(ctx.state, &mut ctx.batch, firehose, *sync)?;
repo_state.root = Some(validated.commit_obj.into());
repo_state.touch();
ctx.stage_repo_state(repo_state);
Ok(Handled::Done)
}
pub(crate) fn handle_identity(
ctx: &mut WorkerContext,
mut repo_state: RepoState<'static>,
firehose: &Url,
mut identity: Box<Identity>,
is_pds: bool,
doc_refreshed: bool,
authority_identity_moved: bool,
) -> Result<Handled<Box<Identity>>> {
let event_ms = identity.time.0.timestamp_millis();
let event_is_new = repo_state.should_process_identity_time(event_ms);
if !event_is_new && !authority_identity_moved {
debug!("skipping stale/duplicate identity event");
return Ok(Handled::Done);
}
if is_pds {
ctx.state.lookups.clear_for_identity(&identity.did);
}
// refresh did doc if a pds sent this event
// or if there is no handle specified. the doc is the authority, so what
// it rewrites is exactly what consumers have to be told about.
let mut changed = authority_identity_moved;
if !doc_refreshed && (is_pds || identity.handle.is_none()) {
match ctx.fact(&identity.did) {
Fact::Resolved(resolved, at) => {
changed |= repo_state.update_from_resolve(resolved.doc.clone(), at);
}
Fact::Failed => warn!("couldnt fetch identity"),
Fact::Unknown => return Ok(Handled::Wait(LookupKind::Refresh, identity)),
}
} else if !doc_refreshed {
// a relay vouches for the handle, not the key, and a rotation comes with an
// #identity too. so the key stops counting as fresh, and the next commits verify
// with it while a refresh in the background checks it
ctx.state.resolver.invalidate_sync(&identity.did);
repo_state.identity_resolved_at = Some(0);
ctx.refresh_in_background(
&identity.did,
RefreshOrigin {
firehose: firehose.clone(),
seq: identity.seq,
time: identity.time.clone(),
},
);
}
// don't pass handle through if it doesnt match ours for pds events
if is_pds && repo_state.handle != identity.handle {
identity.handle = None;
}
if event_is_new {
repo_state.advance_identity_time(event_ms);
}
ctx.sink.identity(
ctx.state,
&mut ctx.batch,
firehose,
*identity,
changed,
true,
)?;
ctx.stage_repo_state(repo_state);
Ok(Handled::Done)
}
pub(crate) fn handle_account(
ctx: &mut WorkerContext,
mut repo_state: RepoState<'static>,
firehose: &Url,
account: Account<'static>,
_is_pds: bool,
) -> Result<()> {
let event_ms = account.time.0.timestamp_millis();
if !repo_state.should_process_account_time(event_ms) {
debug!("skipping stale/duplicate account event");
return Ok(());
}
repo_state.advance_account_time(event_ms);
let was_active = repo_state.active;
let snapshot = ctx.sink.account_snapshot(&repo_state);
repo_state.active = account.active;
if !account.active {
match &account.status {
Some(AccountStatus::Deleted) => {
// keep a Deleted tombstone so any stale commits that arrive later
// (e.g. from the upstream backfill window) are not forwarded.
// per spec: "if any further #commit messages are emitted for the repo,
// all downstream services should ignore the event and not pass it through."
repo_state.status = RepoStatus::Deleted;
}
status => {
repo_state.status = ctx.inactive_account_repo_status(&account.did, status);
}
}
} else {
// active=true: desynchronized/throttled may still carry active=true per spec.
// anything else (including unknown statuses) is treated as synced.
repo_state.status = match &account.status {
Some(AccountStatus::Desynchronized) => RepoStatus::Desynchronized,
Some(AccountStatus::Throttled) => RepoStatus::Throttled,
_ => RepoStatus::Synced,
};
}
ctx.sink.account(
ctx.state,
&mut ctx.batch,
firehose,
account,
&repo_state,
snapshot,
was_active,
)?;
repo_state.touch();
ctx.stage_repo_state(repo_state);
Ok(())
}
}