From 5ecff157aa4507f5b4bf2501c2c14372132a8302 Mon Sep 17 00:00:00 2001
From: Marshall Bowers <git@maxdeviant.com>
Date: Wed, 29 Jan 2025 11:33:48 -0500
Subject: [PATCH] collab: Add internal `POST /snowflake/events` endpoint
 (#23842)

This PR adds a new internal `POST /snowflake/events` endpoint to collab.

This endpoint is protected with the admin token like our other internal
endpoints.

This endpoint accepts a `SnowflakeRow` in the body and writes it to the
AWS Kinesis stream.

Release Notes:

- N/A
---
 crates/collab/src/api.rs | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/crates/collab/src/api.rs b/crates/collab/src/api.rs
index 2b8b5ac6fb7429afca858762eba0e2d65eacbee1..5762a56cb53a481548fcb0b70b1f9305b5c47199 100644
--- a/crates/collab/src/api.rs
+++ b/crates/collab/src/api.rs
@@ -5,6 +5,7 @@ pub mod extensions;
 pub mod ips_file;
 pub mod slack;
 
+use crate::api::events::SnowflakeRow;
 use crate::{
     auth,
     db::{User, UserId},
@@ -99,6 +100,7 @@ pub fn routes(rpc_server: Arc<rpc::Server>) -> Router<(), Body> {
         .route("/user", get(get_authenticated_user))
         .route("/users/:id/access_tokens", post(create_access_token))
         .route("/rpc_server_snapshot", get(get_rpc_server_snapshot))
+        .route("/snowflake/events", post(write_snowflake_event))
         .merge(billing::router())
         .merge(contributors::router())
         .layer(
@@ -245,3 +247,19 @@ async fn create_access_token(
         encrypted_access_token,
     }))
 }
+
+/// An endpoint that writes a Snowflake event to our event stream.
+///
+/// This endpoint is exposed such that other internal services can write
+/// telemetry events without needing to talk to AWS Kinesis directly.
+async fn write_snowflake_event(
+    Extension(app): Extension<Arc<AppState>>,
+    Json(event): Json<SnowflakeRow>,
+) -> Result<()> {
+    let kinesis_client = app.kinesis_client.clone();
+    let kinesis_stream = app.config.kinesis_stream.clone();
+
+    event.write(&kinesis_client, &kinesis_stream).await?;
+
+    Ok(())
+}