merge: #3455

3455: feat(si-layer): basic activities support r=adamhjk a=adamhjk

This adds basic activities into layerdb. Activities represent events that we expect someone else to react to. The first two examples are a quick port of the rebaser events.

```rust
let stream = layerdb.activity_subscribe(Some([vec![ActivityPayloadDiscriminates::RebaseFinished])).await?;

layerdb.activity_publish(&rebase_request)?;
layerdb.activity_publish(&rebase_finished)?;
```

The resulting stream would only receive the second activity event. If you want to subscribe to all events, pass 'None' in as the argument to activity_subscribe.

<img src="https://media0.giphy.com/media/lvDdZh6PZMzvSAUaPj/giphy.gif"/>

Co-authored-by: Adam Jacob <adam@systeminit.com>

Author: si-bors-ng[bot]

Cargo.lock

@@ -83,7 +83,7 @@ impl FromStr for ChangeSetId {
     }
 }
 
-#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
 pub struct Tenancy {
     pub change_set_id: ChangeSetId,
     pub workspace_pk: WorkspacePk,

lib/si-events-rs/src/tenancy.rs

@@ -9,7 +9,6 @@ rust_library(
         "//lib/si-std:si-std",
         "//lib/telemetry-rs:telemetry",
         "//third-party/rust:async-trait",
-        "//third-party/rust:base64",
         "//third-party/rust:blake3",
         "//third-party/rust:bytes",
         "//third-party/rust:chrono",
@@ -20,6 +19,7 @@ rust_library(
         "//third-party/rust:refinery",
         "//third-party/rust:remain",
         "//third-party/rust:serde",
+        "//third-party/rust:serde_json",
         "//third-party/rust:sled",
         "//third-party/rust:strum",
         "//third-party/rust:tempfile",

lib/si-layer-cache/BUCK

@@ -7,7 +7,6 @@ publish = false
 
 [dependencies]
 async-trait = { workspace = true }
-base64 = { workspace = true }
 blake3 = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true }
@@ -30,6 +29,7 @@ thiserror = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
 ulid = { workspace = true }
+serde_json = { workspace = true }
 
 [dev_dependencies]
 buck2-resources = { path = "../../lib/buck2-resources" }

lib/si-layer-cache/Cargo.toml

@@ -0,0 +1,200 @@
+use std::{fmt, str::FromStr, sync::Arc};
+
+use serde::{Deserialize, Serialize};
+use si_data_nats::{async_nats::jetstream, NatsClient};
+use strum::EnumDiscriminants;
+use ulid::{Ulid, ULID_LEN};
+
+use crate::{
+    error::LayerDbResult,
+    event::LayeredEventMetadata,
+    nats::{self, subject},
+};
+
+use self::rebase::{RebaseFinished, RebaseRequest};
+
+pub mod rebase;
+
+#[derive(Copy, Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
+pub struct ActivityId(Ulid);
+
+impl ActivityId {
+    pub fn new() -> ActivityId {
+        ActivityId(Ulid::new())
+    }
+
+    pub fn array_to_str<'buf>(&self, buf: &'buf mut [u8; ULID_LEN]) -> &'buf mut str {
+        self.0.array_to_str(buf)
+    }
+
+    pub fn into_inner(self) -> Ulid {
+        self.0
+    }
+}
+
+impl Default for ActivityId {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl FromStr for ActivityId {
+    type Err = ulid::DecodeError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(Self(Ulid::from_str(s)?))
+    }
+}
+
+impl From<ulid::Ulid> for ActivityId {
+    fn from(value: ulid::Ulid) -> Self {
+        Self(value)
+    }
+}
+
+impl fmt::Display for ActivityId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
+pub struct Activity {
+    pub id: ActivityId,
+    pub payload: ActivityPayload,
+    pub metadata: LayeredEventMetadata,
+}
+
+impl Activity {
+    pub fn new(payload: ActivityPayload, metadata: LayeredEventMetadata) -> Activity {
+        Activity {
+            id: ActivityId::new(),
+            payload,
+            metadata,
+        }
+    }
+
+    pub fn rebase(request: RebaseRequest, metadata: LayeredEventMetadata) -> Activity {
+        Activity::new(ActivityPayload::RebaseRequest(request), metadata)
+    }
+
+    pub fn rebase_finished(request: RebaseFinished, metadata: LayeredEventMetadata) -> Activity {
+        Activity::new(ActivityPayload::RebaseFinished(request), metadata)
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, EnumDiscriminants, PartialEq, Eq)]
+pub enum ActivityPayload {
+    RebaseRequest(RebaseRequest),
+    RebaseFinished(RebaseFinished),
+}
+
+impl ActivityPayload {
+    pub fn to_subject(&self) -> String {
+        let discriminate: ActivityPayloadDiscriminants = self.into();
+        discriminate.to_subject()
+    }
+}
+
+impl ActivityPayloadDiscriminants {
+    pub fn to_subject(&self) -> String {
+        match self {
+            ActivityPayloadDiscriminants::RebaseRequest => "rebase.request".to_string(),
+            ActivityPayloadDiscriminants::RebaseFinished => "rebase.finished".to_string(),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct ActivityPublisher {
+    prefix: Option<Arc<str>>,
+    context: jetstream::context::Context,
+}
+
+impl ActivityPublisher {
+    pub fn new(nats_client: &NatsClient) -> ActivityPublisher {
+        let prefix = nats_client.metadata().subject_prefix().map(|s| s.into());
+        let context = jetstream::new(nats_client.as_inner().clone());
+        ActivityPublisher { context, prefix }
+    }
+
+    pub fn prefix(&self) -> Option<&str> {
+        self.prefix.as_deref()
+    }
+
+    pub fn publish(&self, activity: &Activity) -> LayerDbResult<()> {
+        let nats_subject = subject::for_activity(self.prefix(), activity);
+        let nats = self.context.clone();
+        let nats_payload = postcard::to_stdvec(&activity)?;
+        let _nats_join =
+            tokio::spawn(async move { nats.publish(nats_subject, nats_payload.into()).await });
+        Ok(())
+    }
+}
+
+#[allow(dead_code)]
+pub struct ActivitySubscriber {
+    instance_id: Ulid,
+    messages: jetstream::consumer::pull::Stream,
+}
+
+impl ActivitySubscriber {
+    pub async fn new(
+        instance_id: Ulid,
+        nats_client: &NatsClient,
+        to_receive: Option<Vec<ActivityPayloadDiscriminants>>,
+    ) -> LayerDbResult<ActivitySubscriber> {
+        let context = jetstream::new(nats_client.as_inner().clone());
+
+        let activities =
+            nats::layerdb_activities_stream(&context, nats_client.metadata().subject_prefix())
+                .await?
+                .create_consumer(Self::consumer_config(
+                    nats_client.metadata().subject_prefix(),
+                    instance_id,
+                    to_receive,
+                ))
+                .await?
+                .messages()
+                .await?;
+
+        Ok(ActivitySubscriber {
+            instance_id,
+            messages: activities,
+        })
+    }
+
+    pub fn messages(&mut self) -> &mut jetstream::consumer::pull::Stream {
+        &mut self.messages
+    }
+
+    #[inline]
+    fn consumer_config(
+        prefix: Option<&str>,
+        instance_id: Ulid,
+        to_receive: Option<Vec<ActivityPayloadDiscriminants>>,
+    ) -> jetstream::consumer::pull::Config {
+        let name = format!("activity-stream-{instance_id}");
+        let description = format!("activity stream for [{name}]");
+
+        match to_receive {
+            Some(payload_types) => jetstream::consumer::pull::Config {
+                name: Some(name),
+                description: Some(description),
+                deliver_policy: jetstream::consumer::DeliverPolicy::New,
+                filter_subjects: payload_types
+                    .iter()
+                    .map(|t| nats::subject::for_activity_discriminate(prefix, t))
+                    .map(|s| s.to_string())
+                    .collect(),
+                ..Default::default()
+            },
+            None => jetstream::consumer::pull::Config {
+                name: Some(name),
+                description: Some(description),
+                deliver_policy: jetstream::consumer::DeliverPolicy::New,
+                ..Default::default()
+            },
+        }
+    }
+}

lib/si-layer-cache/src/activities.rs

@@ -0,0 +1,74 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use ulid::Ulid;
+
+/// The message that the server receives to perform a rebase.
+#[derive(Debug, Serialize, Deserialize, Copy, Clone, PartialEq, Eq)]
+pub struct RebaseRequest {
+    /// Corresponds to the change set whose pointer is to be updated.
+    pub to_rebase_change_set_id: Ulid,
+    /// Corresponds to the workspace snapshot that will be the "onto" workspace snapshot when
+    /// rebasing the "to rebase" workspace snapshot.
+    pub onto_workspace_snapshot_id: Ulid,
+    /// Derived from the ephemeral or persisted change set that's either the base change set, the
+    /// last change set before edits were made, or the change set that you are trying to rebase
+    /// onto base.
+    pub onto_vector_clock_id: Ulid,
+}
+
+impl RebaseRequest {
+    pub fn new(
+        to_rebase_change_set_id: Ulid,
+        onto_workspace_snapshot_id: Ulid,
+        onto_vector_clock_id: Ulid,
+    ) -> RebaseRequest {
+        RebaseRequest {
+            to_rebase_change_set_id,
+            onto_workspace_snapshot_id,
+            onto_vector_clock_id,
+        }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub struct RebaseFinished {
+    status: RebaseStatus,
+    to_rebase_change_set_id: Ulid,
+    onto_workspace_snapshot_id: Ulid,
+}
+
+impl RebaseFinished {
+    pub fn new(
+        status: RebaseStatus,
+        to_rebase_change_set_id: Ulid,
+        onto_workspace_snapshot_id: Ulid,
+    ) -> RebaseFinished {
+        RebaseFinished {
+            status,
+            to_rebase_change_set_id,
+            onto_workspace_snapshot_id,
+        }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub enum RebaseStatus {
+    /// Processing the request and performing updates were both successful. Additionally, no conflicts were found.
+    Success {
+        /// The serialized updates performed when rebasing.
+        updates_performed: Value,
+    },
+    /// Conflicts found when processing the request.
+    ConflictsFound {
+        /// A serialized list of the conflicts found during detection.
+        conflicts_found: Value,
+        /// A serialized list of the updates found during detection and skipped because at least
+        /// once conflict was found.
+        updates_found_and_skipped: Value,
+    },
+    /// Error encountered when processing the request.
+    Error {
+        /// The error message.
+        message: String,
+    },
+}

lib/si-layer-cache/src/activities/rebase.rs

@@ -232,7 +232,6 @@ impl ChunkedMessagesStream {
                                 },
                                 ackers: Arc::new(ackers),
                             };
-
                             Some(Poll::Ready(Some(Ok(new_msg))))
                         } else {
                             // Still more chunks to go with this message, so try for more

lib/si-layer-cache/src/chunking_nats.rs

@@ -6,6 +6,7 @@ use telemetry::tracing::{info, warn};
 use ulid::Ulid;
 
 use crate::{
+    activities::{Activity, ActivityPayloadDiscriminants, ActivityPublisher, ActivitySubscriber},
     error::LayerDbResult,
     layer_cache::LayerCache,
     persister::{PersisterClient, PersisterServer},
@@ -28,6 +29,7 @@ where
     pg_pool: PgPool,
     nats_client: NatsClient,
     persister_client: PersisterClient,
+    activity_publisher: ActivityPublisher,
     instance_id: Ulid,
     tracker: TaskTracker,
     cancellation_token: CancellationToken,
@@ -88,7 +90,10 @@ where
 
         let cas = CasDb::new(cas_cache, persister_client.clone());
 
+        let activity_publisher = ActivityPublisher::new(&nats_client);
+
         Ok(LayerDb {
+            activity_publisher,
             cas,
             sled,
             pg_pool,
@@ -138,4 +143,18 @@ where
 
         Ok(())
     }
+
+    // Publish an activity
+    pub fn activity_publish(&self, activity: &Activity) -> LayerDbResult<()> {
+        self.activity_publisher.publish(activity)
+    }
+
+    // Subscribe to all activities, or provide an optional array of activity kinds
+    // to subscribe to.
+    pub async fn activity_subscribe(
+        &self,
+        to_receive: Option<Vec<ActivityPayloadDiscriminants>>,
+    ) -> LayerDbResult<ActivitySubscriber> {
+        ActivitySubscriber::new(self.instance_id, &self.nats_client, to_receive).await
+    }
 }

lib/si-layer-cache/src/db.rs

@@ -8,15 +8,15 @@ use si_events::{Actor, Tenancy, WebEvent};
 use strum::AsRefStr;
 use ulid::Ulid;
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub struct LayeredEventMetadata {
     pub tenancy: Tenancy,
     pub actor: Actor,
     pub timestamp: DateTime<Utc>,
 }
 
 impl LayeredEventMetadata {
-    fn new(tenancy: Tenancy, actor: Actor) -> Self {
+    pub fn new(tenancy: Tenancy, actor: Actor) -> Self {
         LayeredEventMetadata {
             tenancy,
             actor,