use bytes::Bytes;
use jacquard_common::{
CowStr,
types::{
cid::CidLink,
string::{Did, Handle, Tid},
},
};
use serde::{Deserialize, Serialize};
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
#[serde(try_from = "String")]
pub struct Datetime(pub chrono::DateTime<chrono::FixedOffset>);
impl TryFrom<String> for Datetime {
type Error = chrono::ParseError;
fn try_from(s: String) -> Result<Self, Self::Error> {
chrono::DateTime::parse_from_rfc3339(&s)
.map(Self)
.or_else(|_| {
// no timezone, warn and assume UTC
tracing::warn!(
value = %s,
"datetime missing timezone suffix, assuming UTC"
);
chrono::NaiveDateTime::parse_from_str(&s, "%Y-%m-%dT%H:%M:%S%.f")
.map(|ndt| Self(ndt.and_utc().fixed_offset()))
})
}
}
impl jacquard_common::IntoStatic for Datetime {
type Output = Datetime;
fn into_static(self) -> Self::Output {
self
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum RepoOpAction<'a> {
Create,
Update,
Delete,
Other(jacquard_common::CowStr<'a>),
}
impl<'a> RepoOpAction<'a> {
pub fn as_str(&self) -> &str {
match self {
Self::Create => "create",
Self::Update => "update",
Self::Delete => "delete",
Self::Other(s) => s.as_ref(),
}
}
}
impl<'a> From<CowStr<'a>> for RepoOpAction<'a> {
fn from(s: CowStr<'a>) -> Self {
match s.as_ref() {
"create" => Self::Create,
"update" => Self::Update,
"delete" => Self::Delete,
_ => Self::Other(s),
}
}
}
impl<'a> From<&'a str> for RepoOpAction<'a> {
fn from(s: &'a str) -> Self {
Self::from(CowStr::from(s))
}
}
impl<'a> From<String> for RepoOpAction<'a> {
fn from(s: String) -> Self {
Self::from(CowStr::from(s))
}
}
impl<'a> core::fmt::Display for RepoOpAction<'a> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.write_str(self.as_str())
}
}
impl<'a> AsRef<str> for RepoOpAction<'a> {
fn as_ref(&self) -> &str {
self.as_str()
}
}
impl<'a> serde::Serialize for RepoOpAction<'a> {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
serializer.serialize_str(self.as_str())
}
}
impl<'de, 'a> serde::Deserialize<'de> for RepoOpAction<'a>
where
'de: 'a,
{
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
CowStr::deserialize(deserializer).map(Self::from)
}
}
impl<'a> Default for RepoOpAction<'a> {
fn default() -> Self {
Self::Other(Default::default())
}
}
impl<'a> jacquard_common::IntoStatic for RepoOpAction<'a> {
type Output = RepoOpAction<'static>;
fn into_static(self) -> Self::Output {
match self {
RepoOpAction::Create => RepoOpAction::Create,
RepoOpAction::Update => RepoOpAction::Update,
RepoOpAction::Delete => RepoOpAction::Delete,
RepoOpAction::Other(v) => RepoOpAction::Other(v.into_static()),
}
}
}
#[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct RepoOp<'a> {
#[serde(borrow)]
pub action: RepoOpAction<'a>,
/// For creates and updates, the new record CID. For deletions, null.
pub cid: Option<CidLink>,
#[serde(borrow)]
pub path: jacquard_common::CowStr<'a>,
/// For updates and deletes, the previous record CID (required for inductive firehose). For creations, field should not be defined.
#[serde(skip_serializing_if = "std::option::Option::is_none")]
#[serde(default)]
pub prev: Option<CidLink>,
}
impl jacquard_common::IntoStatic for RepoOp<'_> {
type Output = RepoOp<'static>;
fn into_static(self) -> Self::Output {
RepoOp {
action: self.action.into_static(),
cid: self.cid.into_static(),
path: self.path.into_static(),
prev: self.prev.into_static(),
}
}
}
// some relays send `""` for `since` when there is no previous revision instead of null
pub(crate) fn deserialize_tid_or_empty<'de, D>(deserializer: D) -> Result<Option<Tid>, D::Error>
where
D: serde::Deserializer<'de>,
{
let s = <Option<String>>::deserialize(deserializer)?;
match s.as_deref() {
None => Ok(None),
Some("") => {
tracing::warn!("received since with empty string instead of null");
Ok(None)
}
Some(s) => match s.parse() {
Ok(t) => Ok(Some(t)),
Err(_) => {
tracing::warn!(since = s, "received since with invalid tid");
Ok(None)
}
},
}
}
#[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Commit<'a> {
pub blobs: Vec<CidLink>,
#[serde(with = "jacquard_common::serde_bytes_helper")]
pub blocks: Bytes,
pub commit: CidLink,
#[serde(borrow)]
pub ops: Vec<RepoOp<'a>>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
pub prev_data: Option<CidLink>,
#[serde(default)]
pub rebase: bool,
pub repo: Did,
pub rev: Tid,
pub seq: i64,
#[serde(deserialize_with = "deserialize_tid_or_empty")]
pub since: Option<Tid>,
pub time: Datetime,
#[serde(default)]
pub too_big: bool,
}
impl jacquard_common::IntoStatic for Commit<'_> {
type Output = Commit<'static>;
fn into_static(self) -> Self::Output {
Commit {
blobs: self.blobs.into_static(),
blocks: self.blocks,
commit: self.commit.into_static(),
ops: self.ops.into_static(),
prev_data: self.prev_data.into_static(),
rebase: self.rebase,
repo: self.repo.into_static(),
rev: self.rev,
seq: self.seq,
since: self.since,
time: self.time,
too_big: self.too_big,
}
}
}
#[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Identity {
pub did: Did,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
pub handle: Option<Handle>,
pub seq: i64,
pub time: Datetime,
}
impl jacquard_common::IntoStatic for Identity {
type Output = Identity;
fn into_static(self) -> Self::Output {
Identity {
did: self.did.into_static(),
handle: self.handle.into_static(),
seq: self.seq,
time: self.time,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum AccountStatus<'a> {
Takendown,
Suspended,
Deleted,
Deactivated,
Desynchronized,
Throttled,
Other(jacquard_common::CowStr<'a>),
}
impl<'a> AccountStatus<'a> {
pub fn as_str(&self) -> &str {
match self {
Self::Takendown => "takendown",
Self::Suspended => "suspended",
Self::Deleted => "deleted",
Self::Deactivated => "deactivated",
Self::Desynchronized => "desynchronized",
Self::Throttled => "throttled",
Self::Other(s) => s.as_ref(),
}
}
}
impl<'a> From<CowStr<'a>> for AccountStatus<'a> {
fn from(s: CowStr<'a>) -> Self {
match s.as_ref() {
"takendown" => Self::Takendown,
"suspended" => Self::Suspended,
"deleted" => Self::Deleted,
"deactivated" => Self::Deactivated,
"desynchronized" => Self::Desynchronized,
"throttled" => Self::Throttled,
_ => Self::Other(s),
}
}
}
impl<'a> From<&'a str> for AccountStatus<'a> {
fn from(s: &'a str) -> Self {
Self::from(CowStr::from(s))
}
}
impl<'a> From<String> for AccountStatus<'a> {
fn from(s: String) -> Self {
Self::from(CowStr::from(s))
}
}
impl<'a> core::fmt::Display for AccountStatus<'a> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.write_str(self.as_str())
}
}
impl<'a> AsRef<str> for AccountStatus<'a> {
fn as_ref(&self) -> &str {
self.as_str()
}
}
impl<'a> serde::Serialize for AccountStatus<'a> {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
serializer.serialize_str(self.as_str())
}
}
impl<'de, 'a> serde::Deserialize<'de> for AccountStatus<'a>
where
'de: 'a,
{
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
CowStr::deserialize(deserializer).map(Self::from)
}
}
impl<'a> Default for AccountStatus<'a> {
fn default() -> Self {
Self::Other(Default::default())
}
}
impl jacquard_common::IntoStatic for AccountStatus<'_> {
type Output = AccountStatus<'static>;
fn into_static(self) -> Self::Output {
match self {
AccountStatus::Takendown => AccountStatus::Takendown,
AccountStatus::Suspended => AccountStatus::Suspended,
AccountStatus::Deleted => AccountStatus::Deleted,
AccountStatus::Deactivated => AccountStatus::Deactivated,
AccountStatus::Desynchronized => AccountStatus::Desynchronized,
AccountStatus::Throttled => AccountStatus::Throttled,
AccountStatus::Other(v) => AccountStatus::Other(v.into_static()),
}
}
}
#[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Account<'a> {
pub active: bool,
pub did: Did,
pub seq: i64,
#[serde(borrow)]
pub status: Option<AccountStatus<'a>>,
pub time: Datetime,
}
impl jacquard_common::IntoStatic for Account<'_> {
type Output = Account<'static>;
fn into_static(self) -> Self::Output {
Account {
active: self.active,
did: self.did.into_static(),
seq: self.seq,
status: self.status.into_static(),
time: self.time,
}
}
}
#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Sync<'a> {
#[serde(with = "jacquard_common::serde_bytes_helper")]
pub blocks: Bytes,
pub did: Did,
#[serde(borrow)]
pub rev: CowStr<'a>,
pub seq: i64,
pub time: Datetime,
}
impl jacquard_common::IntoStatic for Sync<'_> {
type Output = Sync<'static>;
fn into_static(self) -> Self::Output {
Sync {
blocks: self.blocks,
did: self.did.into_static(),
rev: self.rev.into_static(),
seq: self.seq,
time: self.time,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum InfoName<'a> {
OutdatedCursor,
Other(jacquard_common::CowStr<'a>),
}
impl<'a> InfoName<'a> {
pub fn as_str(&self) -> &str {
match self {
Self::OutdatedCursor => "OutdatedCursor",
Self::Other(s) => s.as_ref(),
}
}
}
impl<'a> From<CowStr<'a>> for InfoName<'a> {
fn from(s: CowStr<'a>) -> Self {
match s.as_ref() {
"OutdatedCursor" => Self::OutdatedCursor,
_ => Self::Other(s),
}
}
}
impl<'a> From<&'a str> for InfoName<'a> {
fn from(s: &'a str) -> Self {
Self::from(CowStr::from(s))
}
}
impl<'a> From<String> for InfoName<'a> {
fn from(s: String) -> Self {
Self::from(CowStr::from(s))
}
}
impl<'a> core::fmt::Display for InfoName<'a> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.write_str(self.as_str())
}
}
impl<'a> AsRef<str> for InfoName<'a> {
fn as_ref(&self) -> &str {
self.as_str()
}
}
impl<'a> serde::Serialize for InfoName<'a> {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
serializer.serialize_str(self.as_str())
}
}
impl<'de, 'a> serde::Deserialize<'de> for InfoName<'a>
where
'de: 'a,
{
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
CowStr::deserialize(deserializer).map(Self::from)
}
}
impl<'a> Default for InfoName<'a> {
fn default() -> Self {
Self::Other(Default::default())
}
}
impl<'a> jacquard_common::IntoStatic for InfoName<'a> {
type Output = InfoName<'static>;
fn into_static(self) -> Self::Output {
match self {
InfoName::OutdatedCursor => InfoName::OutdatedCursor,
InfoName::Other(v) => InfoName::Other(v.into_static()),
}
}
}
#[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Info<'a> {
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(borrow)]
pub message: Option<CowStr<'a>>,
#[serde(borrow)]
pub name: InfoName<'a>,
}
impl jacquard_common::IntoStatic for Info<'_> {
type Output = Info<'static>;
fn into_static(self) -> Self::Output {
Info {
message: self.message.into_static(),
name: self.name.into_static(),
}
}
}
#[derive(Debug, Clone)]
pub enum SubscribeReposMessage<'i> {
Commit(Box<Commit<'i>>),
Sync(Box<Sync<'i>>),
Identity(Box<Identity>),
Account(Box<Account<'i>>),
Info(Box<Info<'i>>),
}
impl jacquard_common::IntoStatic for SubscribeReposMessage<'_> {
type Output = SubscribeReposMessage<'static>;
fn into_static(self) -> Self::Output {
match self {
SubscribeReposMessage::Commit(c) => SubscribeReposMessage::Commit(c.into_static()),
SubscribeReposMessage::Sync(s) => SubscribeReposMessage::Sync(s.into_static()),
SubscribeReposMessage::Identity(i) => SubscribeReposMessage::Identity(i.into_static()),
SubscribeReposMessage::Account(a) => SubscribeReposMessage::Account(a.into_static()),
SubscribeReposMessage::Info(i) => SubscribeReposMessage::Info(i.into_static()),
}
}
}
impl<'i> SubscribeReposMessage<'i> {
pub fn did(&self) -> Option<&Did> {
Some(match self {
SubscribeReposMessage::Commit(c) => &c.repo,
SubscribeReposMessage::Identity(i) => &i.did,
SubscribeReposMessage::Account(a) => &a.did,
SubscribeReposMessage::Sync(s) => &s.did,
_ => return None,
})
}
/// the frame's seq and event time, `None` for an info frame
pub fn position(&self) -> Option<(i64, &Datetime)> {
Some(match self {
SubscribeReposMessage::Commit(c) => (c.seq, &c.time),
SubscribeReposMessage::Identity(i) => (i.seq, &i.time),
SubscribeReposMessage::Account(a) => (a.seq, &a.time),
SubscribeReposMessage::Sync(s) => (s.seq, &s.time),
_ => return None,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Deserialize)]
struct Since {
#[serde(deserialize_with = "deserialize_tid_or_empty")]
since: Option<Tid>,
}
#[test]
fn firehose_scalar_wire_contracts() {
for value in ["create", "update", "delete", "stinkpot"] {
let expected = RepoOpAction::from(value);
let encoded = serde_json::to_vec(&expected).unwrap();
let decoded: RepoOpAction<'_> = serde_json::from_slice(&encoded).unwrap();
assert_eq!(decoded, expected);
assert_eq!(RepoOpAction::from(value.to_owned()), expected);
assert_eq!(expected.as_ref(), value);
assert_eq!(expected.to_string(), value);
assert_eq!(
jacquard_common::IntoStatic::into_static(expected.clone()).as_ref(),
value
);
}
for value in [
"takendown",
"suspended",
"deleted",
"deactivated",
"desynchronized",
"throttled",
"stinkpot",
] {
let expected = AccountStatus::from(value);
let encoded = serde_json::to_vec(&expected).unwrap();
let decoded: AccountStatus<'_> = serde_json::from_slice(&encoded).unwrap();
assert_eq!(decoded, expected);
assert_eq!(AccountStatus::from(value.to_owned()), expected);
assert_eq!(expected.as_ref(), value);
assert_eq!(expected.to_string(), value);
assert_eq!(
jacquard_common::IntoStatic::into_static(expected.clone()).as_ref(),
value
);
}
for value in ["OutdatedCursor", "stinkpot"] {
let expected = InfoName::from(value);
let encoded = serde_json::to_vec(&expected).unwrap();
let decoded: InfoName<'_> = serde_json::from_slice(&encoded).unwrap();
assert_eq!(decoded, expected);
assert_eq!(InfoName::from(value.to_owned()), expected);
assert_eq!(expected.as_ref(), value);
assert_eq!(expected.to_string(), value);
assert_eq!(
jacquard_common::IntoStatic::into_static(expected.clone()).as_ref(),
value
);
}
let action: RepoOpAction<'_> =
RepoOpAction::deserialize(serde::de::value::StringDeserializer::<
serde::de::value::Error,
>::new("stinkpot".to_owned()))
.unwrap();
assert_eq!(action, RepoOpAction::Other("stinkpot".into()));
let status: AccountStatus<'_> =
AccountStatus::deserialize(serde::de::value::StringDeserializer::<
serde::de::value::Error,
>::new("stinkpot".to_owned()))
.unwrap();
assert_eq!(status, AccountStatus::Other("stinkpot".into()));
let name: InfoName<'_> = InfoName::deserialize(serde::de::value::StringDeserializer::<
serde::de::value::Error,
>::new("stinkpot".to_owned()))
.unwrap();
assert_eq!(name, InfoName::Other("stinkpot".into()));
}
#[test]
fn firehose_datetime_accepts_zoned_and_legacy_unzoned_values() {
let zoned: Datetime = serde_json::from_str(r#""2026-08-10T12:34:56.123+02:00""#).unwrap();
assert_eq!(zoned.0.to_rfc3339(), "2026-08-10T12:34:56.123+02:00");
let unzoned: Datetime = serde_json::from_str(r#""2026-08-10T12:34:56.123""#).unwrap();
assert_eq!(unzoned.0.to_rfc3339(), "2026-08-10T12:34:56.123+00:00");
assert_eq!(
jacquard_common::IntoStatic::into_static(zoned.clone()),
zoned
);
assert!(serde_json::from_str::<Datetime>(r#""not-a-datetime""#).is_err());
}
#[test]
fn firehose_since_accepts_null_empty_and_tid_only() {
assert!(
serde_json::from_str::<Since>(r#"{"since":null}"#)
.unwrap()
.since
.is_none()
);
assert!(
serde_json::from_str::<Since>(r#"{"since":""}"#)
.unwrap()
.since
.is_none()
);
let tid = serde_json::from_str::<Since>(r#"{"since":"3jzfcijpj2z2a"}"#)
.unwrap()
.since
.unwrap();
assert_eq!(tid.as_str(), "3jzfcijpj2z2a");
assert!(
serde_json::from_str::<Since>(r#"{"since":"bad"}"#)
.unwrap()
.since
.is_none()
);
}
}