move MAU tracking back to registry

core/startos/src/analytics/mod.rs

@@ -1,27 +1,19 @@
 use std::net::SocketAddr;
 
 use axum::Router;
-use chrono::Utc;
 use futures::future::ready;
-use rpc_toolkit::{from_fn_async, Context, HandlerExt, ParentHandler, Server};
-use sqlx::query;
-use ts_rs::TS;
+use rpc_toolkit::{Context, ParentHandler, Server};
 
 use crate::analytics::context::AnalyticsContext;
 use crate::middleware::cors::Cors;
 use crate::net::static_server::{bad_request, not_found, server_error};
 use crate::net::web_server::WebServer;
 use crate::rpc_continuations::Guid;
-use crate::Error;
 
 pub mod context;
 
 pub fn analytics_api<C: Context>() -> ParentHandler<C> {
     ParentHandler::new()
-        .subcommand(
-            "recordUserActivity",
-            from_fn_async(record_user_activity).no_cli(),
-        )
 }
 
 pub fn analytics_server_router(ctx: AnalyticsContext) -> Router {
@@ -84,29 +76,3 @@ impl WebServer {
         Self::new(bind, analytics_server_router(ctx))
     }
 }
-
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, TS)]
-#[ts(export)]
-#[serde(rename_all = "camelCase")]
-struct ActivityParams {
-    server_id: String,
-    arch: String,
-}
-
-async fn record_user_activity(
-    ctx: AnalyticsContext,
-    ActivityParams { server_id, arch }: ActivityParams,
-) -> Result<(), Error> {
-    let pool = ctx.db;
-    let created_at = Utc::now().to_rfc3339();
-
-    query!("INSERT INTO user_activity (created_at, server_id, os_version, arch) VALUES ($1, $2, $3, $4)",
-    created_at,
-    server_id,
-    arch
-    )
-    .execute(pool)
-    .await?;
-
-    Ok(())
-}

core/startos/src/registry/context.rs

@@ -10,6 +10,7 @@ use reqwest::{Client, Proxy};
 use rpc_toolkit::yajrc::RpcError;
 use rpc_toolkit::{CallRemote, Context, Empty};
 use serde::{Deserialize, Serialize};
+use sqlx::PgPool;
 use tokio::sync::broadcast::Sender;
 use tracing::instrument;
 use url::Url;
@@ -37,6 +38,8 @@ pub struct RegistryConfig {
     pub tor_proxy: Option<Url>,
     #[arg(short = 'd', long = "datadir")]
     pub datadir: Option<PathBuf>,
+    #[arg(short = 'u', long = "pg-connection-url")]
+    pub pg_connection_url: Option<String>,
 }
 impl ContextConfig for RegistryConfig {
     fn next(&mut self) -> Option<PathBuf> {
@@ -67,6 +70,7 @@ pub struct RegistryContextSeed {
     pub rpc_continuations: RpcContinuations,
     pub client: Client,
     pub shutdown: Sender<()>,
+    pub pool: Option<PgPool>,
 }
 
 #[derive(Clone)]
@@ -94,6 +98,13 @@ impl RegistryContext {
             .clone()
             .map(Ok)
             .unwrap_or_else(|| "socks5h://localhost:9050".parse())?;
+        let pool: Option<PgPool> = match &config.pg_connection_url {
+            Some(url) => match PgPool::connect(url.as_str()).await {
+                Ok(pool) => Some(pool),
+                Err(_) => None,
+            },
+            None => None,
+        };
         Ok(Self(Arc::new(RegistryContextSeed {
             hostname: config
                 .hostname
@@ -122,6 +133,7 @@ impl RegistryContext {
                 .build()
                 .with_kind(crate::ErrorKind::ParseUrl)?,
             shutdown,
+            pool,
         })))
     }
 }

core/startos/src/registry/os/version/mod.rs

@@ -1,10 +1,12 @@
 use std::collections::BTreeMap;
 
+use chrono::Utc;
 use clap::Parser;
 use emver::VersionRange;
 use itertools::Itertools;
 use rpc_toolkit::{from_fn_async, Context, HandlerExt, ParentHandler};
 use serde::{Deserialize, Serialize};
+use sqlx::query;
 use ts_rs::TS;
 
 use crate::context::CliContext;
@@ -126,12 +128,34 @@ pub struct GetVersionParams {
     #[ts(type = "string | null")]
     #[arg(long = "target")]
     pub target: Option<VersionRange>,
+    #[ts(type = "string | null")]
+    #[arg(long = "id")]
+    server_id: Option<String>,
+    #[ts(type = "string | null")]
+    #[arg(long = "arch")]
+    arch: Option<String>,
 }
 
 pub async fn get_version(
     ctx: RegistryContext,
-    GetVersionParams { source, target }: GetVersionParams,
+    GetVersionParams {
+        source,
+        target,
+        server_id,
+        arch,
+    }: GetVersionParams,
 ) -> Result<BTreeMap<VersionString, OsVersionInfo>, Error> {
+    if let (Some(pool), Some(server_id), Some(arch)) = (ctx.pool, server_id, arch) {
+        let created_at = Utc::now().to_rfc3339();
+
+        query!("INSERT INTO user_activity (created_at, server_id, os_version, arch) VALUES ($1, $2, $3, $4)",
+        created_at,
+        server_id,
+        arch
+        )
+        .execute(pool)
+        .await?;
+    }
     let target = target.unwrap_or(VersionRange::Any);
     ctx.db
         .peek()