Add Authenticate to Application API (#26450)

This allows us to remove application from sync worker.

GitOrigin-RevId: fdd652e19d3adcafdde17f93b13b9c405b4f589c

Author: Preslav Le

crates/application/src/api.rs

@@ -23,6 +23,7 @@ use futures::{
     future::BoxFuture,
     FutureExt,
 };
+use keybroker::Identity;
 use model::session_requests::types::SessionRequestIdentifier;
 use serde_json::Value as JsonValue;
 use sync_types::{
@@ -58,6 +59,13 @@ pub enum ExecuteQueryTimestamp {
 // version of Convex.
 #[async_trait]
 pub trait ApplicationApi: Send + Sync {
+    async fn authenticate(
+        &self,
+        host: &str,
+        request_id: RequestId,
+        auth_token: AuthenticationToken,
+    ) -> anyhow::Result<Identity>;
+
     async fn execute_public_query(
         &self,
         host: &str,
@@ -104,6 +112,16 @@ pub trait ApplicationApi: Send + Sync {
 // Implements ApplicationApi via Application.
 #[async_trait]
 impl<RT: Runtime> ApplicationApi for Application<RT> {
+    async fn authenticate(
+        &self,
+        _host: &str,
+        _request_id: RequestId,
+        auth_token: AuthenticationToken,
+    ) -> anyhow::Result<Identity> {
+        let validate_time = self.runtime().system_time();
+        self.authenticate(auth_token, validate_time).await
+    }
+
     async fn execute_public_query(
         &self,
         _host: &str,

crates/sync/src/worker.rs

@@ -193,8 +193,6 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
 
 pub struct SyncWorker<RT: Runtime> {
     api: Arc<dyn ApplicationApi>,
-    // TODO(presley): Delete application once all functionality is migrated to api.
-    application: Application<RT>,
     config: SyncWorkerConfig,
     rt: RT,
     state: SyncState,
@@ -243,15 +241,14 @@ impl<RT: Runtime> SyncWorker<RT> {
         rx: UnboundedReceiver<(ClientMessage, RT::Instant)>,
         tx: SingleFlightSender<RT>,
     ) -> Self {
+        let rt = application.runtime().clone();
         // Use api implemented by application until all functionality is migrated
         // and we no longer need application.
-        let api = Arc::new(application.clone());
-        let rt = application.runtime().clone();
+        let api = Arc::new(application);
         let (mutation_sender, receiver) = mpsc::channel(OPERATION_QUEUE_BUFFER_SIZE);
         let mutation_futures = receiver.buffered(1); // Execute at most one operation at a time.
         SyncWorker {
             api,
-            application,
             config,
             rt,
             state: SyncState::new(),
@@ -563,8 +560,8 @@ impl<RT: Runtime> SyncWorker<RT> {
                 base_version,
             } => {
                 let identity = self
-                    .application
-                    .authenticate(auth_token, self.rt.system_time())
+                    .api
+                    .authenticate(self.host.as_str(), RequestId::new(), auth_token)
                     .await?;
                 self.state.modify_identity(identity, base_version)?;
                 self.schedule_update();