diff --git a/.github/services-config.json b/.github/services-config.json index ab5e232bee8..210d6eb52f4 100644 --- a/.github/services-config.json +++ b/.github/services-config.json @@ -35,6 +35,16 @@ "upload_extractor_lambda_trigger" ] }, + "calendar-event-local-tunnel": { + "source_paths": [ + "services/calendar_event_local_tunnel/**", + "crates/calendar_watch_relay/**" + ], + "stack_path": "infra/stacks/calendar-event-local-tunnel/**", + "deploy_binaries": [ + "calendar_event_local_tunnel" + ] + }, "call-recording": { "source_paths": [ "services/call_recording_preview_handler/**" diff --git a/.github/workflows/deploy-service-generic.yml b/.github/workflows/deploy-service-generic.yml index ff5ab7c6780..03434486807 100644 --- a/.github/workflows/deploy-service-generic.yml +++ b/.github/workflows/deploy-service-generic.yml @@ -10,6 +10,7 @@ on: - agent-schedule-service - authentication-service - bulk-upload + - calendar-event-local-tunnel - connection-gateway - contacts-service - convert-service diff --git a/.github/workspace-dep-closures.json b/.github/workspace-dep-closures.json index 4d5bd02cda8..0772f4eb346 100644 --- a/.github/workspace-dep-closures.json +++ b/.github/workspace-dep-closures.json @@ -606,6 +606,17 @@ "crates/model-error-response", "crates/workspace-hack" ], + "calendar_event_local_tunnel": [ + "crates/calendar_watch_relay", + "crates/macro_config", + "crates/macro_config_derive", + "crates/macro_entrypoint", + "crates/macro_env", + "crates/macro_env_var", + "crates/remote_env_var", + "crates/workspace-hack", + "services/calendar_event_local_tunnel" + ], "calendar_events": [ "crates/ai_toolset", "crates/bot_id", @@ -635,6 +646,11 @@ "crates/remote_env_var", "crates/workspace-hack" ], + "calendar_watch_relay": [ + "crates/calendar_watch_relay", + "crates/macro_env_var", + "crates/workspace-hack" + ], "call": [ "crates/activity", "crates/agent", @@ -2294,6 +2310,7 @@ "crates/authentication_service_client", "crates/bot_id", "crates/calendar_events", + "crates/calendar_watch_relay", "crates/channel_sender", "crates/connection_gateway_client", "crates/connection_gateway_models", diff --git a/.sqlx/query-a6ea220aaff4d5b84963280a24d51034b8f16fc6dbd288ebbd027cfde86b7872.json b/.sqlx/query-a6ea220aaff4d5b84963280a24d51034b8f16fc6dbd288ebbd027cfde86b7872.json new file mode 100644 index 00000000000..f3eb2ac1f60 --- /dev/null +++ b/.sqlx/query-a6ea220aaff4d5b84963280a24d51034b8f16fc6dbd288ebbd027cfde86b7872.json @@ -0,0 +1,56 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n calendar.id AS calendar_id,\n calendar.watch_channel_id AS \"channel_id!\",\n calendar.watch_resource_id AS \"resource_id!\",\n account.email_link_id,\n link.fusionauth_user_id,\n link.email_address,\n link.provider::text AS \"provider!\"\n FROM calendars calendar\n JOIN calendar_accounts account ON account.id = calendar.account_id\n JOIN email_links link ON link.id = account.email_link_id\n WHERE calendar.watch_channel_id IS NOT NULL\n AND calendar.watch_resource_id IS NOT NULL\n AND (calendar.watch_expires_at IS NULL OR calendar.watch_expires_at > now())\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "calendar_id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "channel_id!", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "resource_id!", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "email_link_id", + "type_info": "Uuid" + }, + { + "ordinal": 4, + "name": "fusionauth_user_id", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "email_address", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "provider!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + true, + true, + false, + false, + false, + null + ] + }, + "hash": "a6ea220aaff4d5b84963280a24d51034b8f16fc6dbd288ebbd027cfde86b7872" +} diff --git a/.sqlx/query-ec4d63f3ed765a0f289331e66a95171e097fa6ff3676cc64fdcbebf517bea83b.json b/.sqlx/query-ec4d63f3ed765a0f289331e66a95171e097fa6ff3676cc64fdcbebf517bea83b.json new file mode 100644 index 00000000000..7f69edeca10 --- /dev/null +++ b/.sqlx/query-ec4d63f3ed765a0f289331e66a95171e097fa6ff3676cc64fdcbebf517bea83b.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE calendars\n SET watch_channel_id = NULL,\n watch_resource_id = NULL,\n watch_expires_at = NULL,\n updated_at = now()\n WHERE id = $1\n AND watch_channel_id = $2\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "ec4d63f3ed765a0f289331e66a95171e097fa6ff3676cc64fdcbebf517bea83b" +} diff --git a/Cargo.lock b/Cargo.lock index 66677f28251..1b551a11337 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2551,6 +2551,28 @@ dependencies = [ "workspace-hack", ] +[[package]] +name = "calendar_event_local_tunnel" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum", + "calendar_watch_relay", + "doppler-config", + "futures", + "macro_config", + "macro_entrypoint", + "macro_env", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "tower 0.5.3", + "tower-http", + "tracing", + "workspace-hack", +] + [[package]] name = "calendar_events" version = "0.1.0" @@ -2581,6 +2603,17 @@ dependencies = [ "workspace-hack", ] +[[package]] +name = "calendar_watch_relay" +version = "0.1.0" +dependencies = [ + "macro_env_var", + "serde", + "serde_json", + "sha2 0.10.9", + "workspace-hack", +] + [[package]] name = "call" version = "0.1.0" @@ -5193,6 +5226,7 @@ dependencies = [ "base64 0.22.1", "bytes", "calendar_events", + "calendar_watch_relay", "chrono", "cloudfront_sign", "connection_gateway_client", @@ -15131,6 +15165,7 @@ dependencies = [ "futures-core", "pin-project-lite", "tokio", + "tokio-util", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 89f525ac47b..4ccbb179a66 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,7 @@ members = [ "crates/bots", "crates/broadcast", "crates/calendar_events", + "crates/calendar_watch_relay", "crates/channel_bots", "crates/client/cache-core", "crates/client/cache-idb", @@ -60,6 +61,7 @@ members = [ "crates/workspace-hack", "services/ai_projections_refresh_handler", "services/authentication_service", + "services/calendar_event_local_tunnel", "services/call_recording_preview_handler", "services/connection_gateway", "services/contacts_service", diff --git a/crates/calendar_events/src/domain/models.rs b/crates/calendar_events/src/domain/models.rs index d63dbe10f40..be66e07176e 100644 --- a/crates/calendar_events/src/domain/models.rs +++ b/crates/calendar_events/src/domain/models.rs @@ -866,6 +866,32 @@ pub struct GoogleWatchChannel { pub expires_at: DateTime, } +/// Outcome of a best-effort pass stopping every open push channel. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct WatchChannelStopSummary { + /// Channels confirmed stopped (or already gone) and cleared. + pub stopped: usize, + /// Channels not fully torn down this pass — the stop call or its + /// bookkeeping cleanup failed. Their bookkeeping is kept so a later pass + /// or natural expiry finishes the job. + pub failed: usize, +} + +/// An open push notification channel joined to the identity able to stop it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ActiveWatchChannel { + /// Calendar row holding the channel bookkeeping. + pub calendar_id: Uuid, + /// Client-minted channel identifier. + pub channel_id: String, + /// Provider-assigned resource identifier. + pub resource_id: String, + /// Link whose grant opened the channel (also the request-gate key). + pub email_link_id: Uuid, + /// Refresh-token identity used to mint an access token for the stop call. + pub token_identity: CalendarLinkTokenIdentity, +} + /// How the provider adapter must reconcile one calendar this run. #[derive(Clone, Debug, PartialEq, Eq)] pub enum GoogleSyncPlan { diff --git a/crates/calendar_events/src/domain/ports.rs b/crates/calendar_events/src/domain/ports.rs index 1bcc31b0423..51d4279d412 100644 --- a/crates/calendar_events/src/domain/ports.rs +++ b/crates/calendar_events/src/domain/ports.rs @@ -7,7 +7,7 @@ use rootcause::Report; use uuid::Uuid; use super::models::{ - AppliedGoogleGrant, AttendeeResponseStatus, CalendarBackfillClaim, + ActiveWatchChannel, AppliedGoogleGrant, AttendeeResponseStatus, CalendarBackfillClaim, CalendarBackfillFailureDisposition, CalendarBackfillFailureOutcome, CalendarBackfillJobKey, CalendarCreationTarget, CalendarEvent, CalendarEventDraft, CalendarEventMutationTarget, CalendarEventPatch, CalendarEventUpsert, CalendarLinkTokenIdentity, CalendarOccurrence, @@ -483,6 +483,35 @@ pub trait GoogleCalendarProvider: Send + Sync + 'static { ) -> impl Future> + Send; } +/// Provider-side teardown of push notification channels, kept separate from +/// [`GoogleCalendarProvider`] because only explicit channel teardown needs it. +pub trait GoogleWatchChannelStopper: Send + Sync + 'static { + /// Stop one push channel. An already-gone channel counts as success. + fn stop_watch_channel( + &self, + access_token: &str, + email_link_id: Uuid, + channel_id: &str, + resource_id: &str, + ) -> impl Future> + Send; +} + +/// Bookkeeping needed to stop every open push channel at teardown. +pub trait WatchChannelTeardownRepository: Send + Sync + 'static { + /// List every unexpired push channel with the identity that can stop it. + fn list_active_watch_channels( + &self, + ) -> impl Future, Report>> + Send; + + /// Clear one channel's bookkeeping, guarded by channel id so a + /// concurrently reopened channel is never clobbered. + fn clear_watch_channel( + &self, + calendar_id: Uuid, + channel_id: &str, + ) -> impl Future> + Send; +} + /// Durable scheduling operations for periodic provider maintenance. pub trait GoogleCalendarSyncRepository: Send + Sync + 'static { /// Reset completed current-grant jobs that are due for another incremental poll. diff --git a/crates/calendar_events/src/domain/service.rs b/crates/calendar_events/src/domain/service.rs index b387944602f..aac02d54401 100644 --- a/crates/calendar_events/src/domain/service.rs +++ b/crates/calendar_events/src/domain/service.rs @@ -10,12 +10,13 @@ use super::{ AppliedGoogleGrant, CalendarBackfillClaim, CalendarBackfillFailureDisposition, CalendarBackfillFailureOutcome, CalendarBackfillJobKey, CalendarEventUpsert, CalendarOccurrenceCursor, GoogleBackfillRunReport, GoogleCalendarSyncSnapshot, - GoogleScopeSet, OccurrenceRange, + GoogleScopeSet, OccurrenceRange, WatchChannelStopSummary, }, ports::{ - CalendarBackfillRepository, CalendarEventWrite, CalendarOccurrenceService, - CalendarRepository, GoogleCalendarProvider, GoogleCalendarSyncRepository, - GoogleEventSyncContext, GoogleProviderError, GoogleProviderErrorKind, + CalendarAccessTokenProvider, CalendarBackfillRepository, CalendarEventWrite, + CalendarOccurrenceService, CalendarRepository, GoogleCalendarProvider, + GoogleCalendarSyncRepository, GoogleEventSyncContext, GoogleProviderError, + GoogleProviderErrorKind, GoogleWatchChannelStopper, WatchChannelTeardownRepository, }, }; @@ -609,6 +610,92 @@ where } } +/// Best-effort stop of every open push notification channel. +/// +/// Local stacks run this at shutdown so Google stops delivering to a +/// subscription that no longer has a consumer; deployments never need it +/// because their channels are renewed for as long as the deployment lives. +/// Tokens are minted once per link identity, and every failure is logged and +/// skipped: a channel left behind keeps its bookkeeping and simply lapses at +/// its natural expiry. +pub async fn stop_all_watch_channels( + repository: &R, + provider: &G, + tokens: &T, +) -> Result +where + R: WatchChannelTeardownRepository, + G: GoogleWatchChannelStopper, + T: CalendarAccessTokenProvider, +{ + let channels = repository.list_active_watch_channels().await?; + let mut summary = WatchChannelStopSummary::default(); + let mut tokens_by_link: std::collections::HashMap> = + std::collections::HashMap::new(); + for channel in channels { + let access_token = match tokens_by_link.entry(channel.email_link_id) { + std::collections::hash_map::Entry::Occupied(entry) => entry.get().clone(), + std::collections::hash_map::Entry::Vacant(entry) => { + let minted = tokens + .fetch_access_token(&channel.token_identity) + .await + .inspect_err(|error| { + tracing::warn!( + error=?error, + email_link_id=%channel.email_link_id, + "failed to mint an access token to stop watch channels" + ); + }) + .ok(); + entry.insert(minted).clone() + } + }; + let Some(access_token) = access_token else { + summary.failed += 1; + continue; + }; + match provider + .stop_watch_channel( + &access_token, + channel.email_link_id, + &channel.channel_id, + &channel.resource_id, + ) + .await + { + Ok(()) => { + match repository + .clear_watch_channel(channel.calendar_id, &channel.channel_id) + .await + { + Ok(()) => summary.stopped += 1, + // Kept bookkeeping means a later pass retries the whole + // teardown; the stop call tolerates already-gone channels, + // so the retry converges. + Err(error) => { + tracing::warn!( + error=?error, + calendar_id=%channel.calendar_id, + "stopped a watch channel but failed to clear its bookkeeping" + ); + summary.failed += 1; + } + } + } + Err(error) => { + tracing::warn!( + error=?error, + calendar_id=%channel.calendar_id, + channel_id=%channel.channel_id, + "failed to stop a watch channel" + ); + summary.failed += 1; + } + } + } + Ok(summary) +} + fn validate_upsert(upsert: &CalendarEventUpsert) -> Result<(), Report> { if upsert.event.owner_id.trim().is_empty() || upsert.event.ical_uid.trim().is_empty() { return Err(rootcause::report!(CalendarValidationError::MissingIdentity).into()); diff --git a/crates/calendar_events/src/domain/service/test.rs b/crates/calendar_events/src/domain/service/test.rs index dd0f8ed4598..65bac839a4c 100644 --- a/crates/calendar_events/src/domain/service/test.rs +++ b/crates/calendar_events/src/domain/service/test.rs @@ -1,17 +1,17 @@ use super::*; use crate::domain::{ models::{ - AttendeeResponseStatus, CalendarAttendee, CalendarBackfillClaim, + ActiveWatchChannel, AttendeeResponseStatus, CalendarAttendee, CalendarBackfillClaim, CalendarBackfillFailureDisposition, CalendarBackfillJobKey, CalendarCreationTarget, - CalendarEvent, CalendarEventMutationTarget, CalendarEventSource, CalendarOccurrence, - CalendarSyncStatus, EventReminders, EventStatus, EventTime, EventTransparency, - EventVisibility, GOOGLE_CALENDAR_SCOPES, GoogleBackfillRunReport, + CalendarEvent, CalendarEventMutationTarget, CalendarEventSource, CalendarLinkTokenIdentity, + CalendarOccurrence, CalendarSyncStatus, EventReminders, EventStatus, EventTime, + EventTransparency, EventVisibility, GOOGLE_CALENDAR_SCOPES, GoogleBackfillRunReport, GoogleCalendarSyncSnapshot, GoogleEventSource, GoogleEventSyncBatch, GoogleWatchChannel, GoogleWatchConfig, ProviderCalendar, StoredGoogleCalendar, }, ports::{ - CalendarBackfillRepository, CalendarEventWrite, CalendarRepository, GoogleCalendarProvider, - GoogleEventSyncContext, GoogleProviderError, + CalendarBackfillRepository, CalendarEventWrite, CalendarRepository, CalendarTokenError, + GoogleCalendarProvider, GoogleEventSyncContext, GoogleProviderError, }, }; use chrono::{TimeZone, Utc}; @@ -702,3 +702,212 @@ async fn google_coordinator_keeps_calendar_permission_health_separate_from_gmail &[CalendarBackfillFailureDisposition::CalendarPermissionRequired] ); } + +#[derive(Clone, Default)] +struct FakeTeardownRepo { + channels: Vec, + fail_clear_for: Option, + cleared: Arc>>, +} + +impl WatchChannelTeardownRepository for FakeTeardownRepo { + async fn list_active_watch_channels(&self) -> Result, Report> { + Ok(self.channels.clone()) + } + + async fn clear_watch_channel(&self, calendar_id: Uuid, channel_id: &str) -> Result<(), Report> { + if self.fail_clear_for.as_deref() == Some(channel_id) { + return Err(rootcause::report!(CalendarValidationError::MissingIdentity).into()); + } + self.cleared + .lock() + .unwrap() + .push((calendar_id, channel_id.to_owned())); + Ok(()) + } +} + +#[derive(Clone, Default)] +struct FakeStopper { + fail_channel: Option, + stopped: Arc>>, +} + +impl GoogleWatchChannelStopper for FakeStopper { + async fn stop_watch_channel( + &self, + access_token: &str, + _email_link_id: Uuid, + channel_id: &str, + _resource_id: &str, + ) -> Result<(), GoogleProviderError> { + if self.fail_channel.as_deref() == Some(channel_id) { + return Err(GoogleProviderError::new( + GoogleProviderErrorKind::Transient, + "stop refused", + )); + } + self.stopped + .lock() + .unwrap() + .push((access_token.to_owned(), channel_id.to_owned())); + Ok(()) + } +} + +#[derive(Clone, Default)] +struct FakeTokens { + fail_for: Option, + minted: Arc>>, +} + +impl CalendarAccessTokenProvider for FakeTokens { + async fn fetch_access_token( + &self, + identity: &CalendarLinkTokenIdentity, + ) -> Result { + if self.fail_for.as_deref() == Some(identity.email_address.as_str()) { + return Err(CalendarTokenError::Transient("mint refused".to_owned())); + } + self.minted + .lock() + .unwrap() + .push(identity.email_address.clone()); + Ok(format!("token-{}", identity.email_address)) + } +} + +fn active_channel(link_id: Uuid, email: &str, channel_id: &str) -> ActiveWatchChannel { + ActiveWatchChannel { + calendar_id: Uuid::now_v7(), + channel_id: channel_id.to_owned(), + resource_id: format!("resource-{channel_id}"), + email_link_id: link_id, + token_identity: CalendarLinkTokenIdentity { + fusionauth_user_id: format!("fa-{email}"), + email_address: email.to_owned(), + provider: "GMAIL".to_owned(), + }, + } +} + +#[tokio::test] +async fn stop_all_stops_and_clears_every_channel_minting_once_per_link() { + let link_a = Uuid::now_v7(); + let link_b = Uuid::now_v7(); + let repo = FakeTeardownRepo { + channels: vec![ + active_channel(link_a, "a@example.com", "chan-1"), + active_channel(link_a, "a@example.com", "chan-2"), + active_channel(link_b, "b@example.com", "chan-3"), + ], + ..Default::default() + }; + let stopper = FakeStopper::default(); + let tokens = FakeTokens::default(); + + let summary = stop_all_watch_channels(&repo, &stopper, &tokens) + .await + .unwrap(); + + assert_eq!(summary.stopped, 3); + assert_eq!(summary.failed, 0); + assert_eq!( + tokens.minted.lock().unwrap().len(), + 2, + "one token per link, not per channel" + ); + let stopped = stopper.stopped.lock().unwrap(); + assert!( + stopped + .iter() + .any(|(token, channel)| { token == "token-a@example.com" && channel == "chan-1" }) + ); + assert_eq!(repo.cleared.lock().unwrap().len(), 3); +} + +#[tokio::test] +async fn stop_all_keeps_bookkeeping_for_channels_that_fail_to_stop() { + let link = Uuid::now_v7(); + let repo = FakeTeardownRepo { + channels: vec![ + active_channel(link, "a@example.com", "chan-ok"), + active_channel(link, "a@example.com", "chan-bad"), + ], + ..Default::default() + }; + let stopper = FakeStopper { + fail_channel: Some("chan-bad".to_owned()), + ..Default::default() + }; + let tokens = FakeTokens::default(); + + let summary = stop_all_watch_channels(&repo, &stopper, &tokens) + .await + .unwrap(); + + assert_eq!(summary.stopped, 1); + assert_eq!(summary.failed, 1); + let cleared = repo.cleared.lock().unwrap(); + assert_eq!(cleared.len(), 1); + assert_eq!(cleared[0].1, "chan-ok"); +} + +#[tokio::test] +async fn stop_all_counts_a_failed_bookkeeping_clear_as_failed() { + let link = Uuid::now_v7(); + let repo = FakeTeardownRepo { + channels: vec![ + active_channel(link, "a@example.com", "chan-ok"), + active_channel(link, "a@example.com", "chan-uncleared"), + ], + fail_clear_for: Some("chan-uncleared".to_owned()), + ..Default::default() + }; + let stopper = FakeStopper::default(); + let tokens = FakeTokens::default(); + + let summary = stop_all_watch_channels(&repo, &stopper, &tokens) + .await + .unwrap(); + + assert_eq!(summary.stopped, 1); + assert_eq!( + summary.failed, 1, + "a stopped channel whose bookkeeping survives is not fully torn down" + ); + let cleared = repo.cleared.lock().unwrap(); + assert_eq!(cleared.len(), 1); + assert_eq!(cleared[0].1, "chan-ok"); +} + +#[tokio::test] +async fn stop_all_counts_channels_whose_token_cannot_be_minted() { + let failing_link = Uuid::now_v7(); + let healthy_link = Uuid::now_v7(); + let repo = FakeTeardownRepo { + channels: vec![ + active_channel(failing_link, "broken@example.com", "chan-1"), + active_channel(failing_link, "broken@example.com", "chan-2"), + active_channel(healthy_link, "ok@example.com", "chan-3"), + ], + ..Default::default() + }; + let stopper = FakeStopper::default(); + let tokens = FakeTokens { + fail_for: Some("broken@example.com".to_owned()), + ..Default::default() + }; + + let summary = stop_all_watch_channels(&repo, &stopper, &tokens) + .await + .unwrap(); + + assert_eq!(summary.stopped, 1); + assert_eq!(summary.failed, 2); + assert_eq!( + stopper.stopped.lock().unwrap().len(), + 1, + "no stop attempts without a token" + ); +} diff --git a/crates/calendar_events/src/outbound/google.rs b/crates/calendar_events/src/outbound/google.rs index c7ddfbc92f5..199e8b94723 100644 --- a/crates/calendar_events/src/outbound/google.rs +++ b/crates/calendar_events/src/outbound/google.rs @@ -20,7 +20,7 @@ use crate::domain::{ ports::{ CalendarRsvpScope, GoogleCalendarMutationProvider, GoogleCalendarProvider, GoogleEventSyncContext, GoogleProviderError, GoogleProviderErrorKind, GoogleRsvpOutcome, - GoogleSeriesMutationOutcome, + GoogleSeriesMutationOutcome, GoogleWatchChannelStopper, }, }; @@ -528,6 +528,36 @@ impl GoogleCalendarProvider for GoogleCalendarClient { } } +impl GoogleWatchChannelStopper for GoogleCalendarClient { + async fn stop_watch_channel( + &self, + access_token: &str, + email_link_id: Uuid, + channel_id: &str, + resource_id: &str, + ) -> Result<(), GoogleProviderError> { + self.gate.acquire(email_link_id).await?; + let response = self + .client + .post(format!("{GOOGLE_CALENDAR_API}/channels/stop")) + .bearer_auth(access_token) + .json(&serde_json::json!({ + "id": channel_id, + "resourceId": resource_id, + })) + .send() + .await + .map_err(provider_transport_error)?; + let status = response.status(); + // A channel Google no longer knows is already in the desired state. + if status.is_success() || status == StatusCode::NOT_FOUND { + return Ok(()); + } + let body = response.text().await.map_err(provider_transport_error)?; + Err(provider_response_error(status, &body)) + } +} + /// Feed application state shared by incremental polls and tail extension. #[derive(Default)] struct AppliedChangeFeed { diff --git a/crates/calendar_events/src/outbound/pg.rs b/crates/calendar_events/src/outbound/pg.rs index 770384b93ea..8c5f7a91680 100644 --- a/crates/calendar_events/src/outbound/pg.rs +++ b/crates/calendar_events/src/outbound/pg.rs @@ -11,10 +11,10 @@ use uuid::Uuid; use crate::domain::{ models::{ - AppliedGoogleGrant, AttendeeResponseStatus, CalendarAttendee, CalendarBackfillClaim, - CalendarBackfillFailureDisposition, CalendarBackfillFailureOutcome, CalendarBackfillJob, - CalendarBackfillJobKey, CalendarBackfillKind, CalendarCreationTarget, CalendarEvent, - CalendarEventMutationTarget, CalendarEventOverride, CalendarEventSource, + ActiveWatchChannel, AppliedGoogleGrant, AttendeeResponseStatus, CalendarAttendee, + CalendarBackfillClaim, CalendarBackfillFailureDisposition, CalendarBackfillFailureOutcome, + CalendarBackfillJob, CalendarBackfillJobKey, CalendarBackfillKind, CalendarCreationTarget, + CalendarEvent, CalendarEventMutationTarget, CalendarEventOverride, CalendarEventSource, CalendarEventUpsert, CalendarLinkTokenIdentity, CalendarOccurrence, CalendarOccurrenceCursor, CalendarReminderFiring, CalendarSyncStatus, DueCalendarReminder, EventReminderOverride, EventReminders, EventStart, EventStatus, EventTime, @@ -24,7 +24,7 @@ use crate::domain::{ }, ports::{ CalendarBackfillRepository, CalendarEventWrite, CalendarReminderDispatchRepo, - CalendarRepository, GoogleCalendarSyncRepository, + CalendarRepository, GoogleCalendarSyncRepository, WatchChannelTeardownRepository, }, }; @@ -3196,6 +3196,68 @@ impl CalendarReminderDispatchRepo for PgCalendarRepository { } } +impl WatchChannelTeardownRepository for PgCalendarRepository { + #[tracing::instrument(skip(self), err)] + async fn list_active_watch_channels(&self) -> Result, Report> { + let rows = sqlx::query!( + r#" + SELECT + calendar.id AS calendar_id, + calendar.watch_channel_id AS "channel_id!", + calendar.watch_resource_id AS "resource_id!", + account.email_link_id, + link.fusionauth_user_id, + link.email_address, + link.provider::text AS "provider!" + FROM calendars calendar + JOIN calendar_accounts account ON account.id = calendar.account_id + JOIN email_links link ON link.id = account.email_link_id + WHERE calendar.watch_channel_id IS NOT NULL + AND calendar.watch_resource_id IS NOT NULL + AND (calendar.watch_expires_at IS NULL OR calendar.watch_expires_at > now()) + "#, + ) + .fetch_all(&self.pool) + .await + .map_err(report)?; + Ok(rows + .into_iter() + .map(|row| ActiveWatchChannel { + calendar_id: row.calendar_id, + channel_id: row.channel_id, + resource_id: row.resource_id, + email_link_id: row.email_link_id, + token_identity: CalendarLinkTokenIdentity { + fusionauth_user_id: row.fusionauth_user_id, + email_address: row.email_address, + provider: row.provider, + }, + }) + .collect()) + } + + #[tracing::instrument(skip(self, channel_id), err)] + async fn clear_watch_channel(&self, calendar_id: Uuid, channel_id: &str) -> Result<(), Report> { + sqlx::query!( + r#" + UPDATE calendars + SET watch_channel_id = NULL, + watch_resource_id = NULL, + watch_expires_at = NULL, + updated_at = now() + WHERE id = $1 + AND watch_channel_id = $2 + "#, + calendar_id, + channel_id, + ) + .execute(&self.pool) + .await + .map_err(report)?; + Ok(()) + } +} + fn event_reconciliation_lock(source_link_id: Uuid, ical_uid: &str) -> i64 { // Stable FNV-1a produces the same advisory-lock key in every service // process. A collision only adds harmless serialization. diff --git a/crates/calendar_events/src/outbound/pg/test.rs b/crates/calendar_events/src/outbound/pg/test.rs index da282adb97f..e4ae2df1571 100644 --- a/crates/calendar_events/src/outbound/pg/test.rs +++ b/crates/calendar_events/src/outbound/pg/test.rs @@ -2871,3 +2871,156 @@ async fn stale_and_declined_firings_resolve_safely(pool: PgPool) { .unwrap(); assert_eq!(past, Vec::new()); } + +#[sqlx::test(migrator = "MACRO_DB_MIGRATIONS")] +async fn watch_channel_teardown_lists_and_clears_open_channels(pool: PgPool) { + let owner_id = "macro|watch-teardown@example.com"; + let link_id = insert_link(&pool, owner_id).await; + let repo = PgCalendarRepository::new(pool.clone()); + let enabled = repo + .apply_google_grant(link_id, complete_grant()) + .await + .unwrap(); + let account_id = repo.upsert_google_account(link_id).await.unwrap(); + let google_job = enabled + .jobs + .into_iter() + .find(|job| job.kind == CalendarBackfillKind::GoogleCalendar) + .unwrap(); + let key = CalendarBackfillJobKey { + job_id: google_job.id, + email_link_id: link_id, + }; + let CalendarBackfillClaim::Claimed { lease_token, .. } = + repo.claim_google_backfill(key).await.unwrap() + else { + panic!("Google job should be claimable"); + }; + let calendar_id = repo + .upsert_google_calendar( + key, + lease_token, + account_id, + ProviderCalendar { + provider_calendar_id: "primary".to_string(), + name: "Primary".to_string(), + description: None, + time_zone: Some("UTC".to_string()), + color: None, + access_role: Some("owner".to_string()), + is_primary: true, + is_selected: true, + }, + ) + .await + .unwrap() + .id; + let channel = GoogleWatchChannel { + channel_id: Uuid::new_v4(), + resource_id: "resource-1".to_string(), + expires_at: (Utc::now() + Duration::days(6)).trunc_subsecs(6), + }; + repo.record_watch_channel(key, lease_token, account_id, calendar_id, channel.clone()) + .await + .unwrap(); + + let active = repo.list_active_watch_channels().await.unwrap(); + let listed = active + .iter() + .find(|entry| entry.calendar_id == calendar_id) + .expect("the open channel is listed"); + assert_eq!(listed.channel_id, channel.channel_id.to_string()); + assert_eq!(listed.resource_id, "resource-1"); + assert_eq!(listed.email_link_id, link_id); + assert_eq!(listed.token_identity.fusionauth_user_id, owner_id); + assert_eq!(listed.token_identity.provider, "GMAIL"); + + // Clearing is guarded by channel id, so a reopened channel can't be + // clobbered by a stale teardown. + repo.clear_watch_channel(calendar_id, "some-other-channel") + .await + .unwrap(); + assert!( + repo.list_active_watch_channels() + .await + .unwrap() + .iter() + .any(|entry| entry.calendar_id == calendar_id), + "a mismatched channel id must not clear the bookkeeping" + ); + + repo.clear_watch_channel(calendar_id, &channel.channel_id.to_string()) + .await + .unwrap(); + assert!( + !repo + .list_active_watch_channels() + .await + .unwrap() + .iter() + .any(|entry| entry.calendar_id == calendar_id), + "a cleared channel is no longer listed" + ); +} + +#[sqlx::test(migrator = "MACRO_DB_MIGRATIONS")] +async fn watch_channel_teardown_skips_already_expired_channels(pool: PgPool) { + let link_id = insert_link(&pool, "macro|watch-expired@example.com").await; + let repo = PgCalendarRepository::new(pool.clone()); + let enabled = repo + .apply_google_grant(link_id, complete_grant()) + .await + .unwrap(); + let account_id = repo.upsert_google_account(link_id).await.unwrap(); + let google_job = enabled + .jobs + .into_iter() + .find(|job| job.kind == CalendarBackfillKind::GoogleCalendar) + .unwrap(); + let key = CalendarBackfillJobKey { + job_id: google_job.id, + email_link_id: link_id, + }; + let CalendarBackfillClaim::Claimed { lease_token, .. } = + repo.claim_google_backfill(key).await.unwrap() + else { + panic!("Google job should be claimable"); + }; + let calendar_id = repo + .upsert_google_calendar( + key, + lease_token, + account_id, + ProviderCalendar { + provider_calendar_id: "primary".to_string(), + name: "Primary".to_string(), + description: None, + time_zone: Some("UTC".to_string()), + color: None, + access_role: Some("owner".to_string()), + is_primary: true, + is_selected: true, + }, + ) + .await + .unwrap() + .id; + let channel = GoogleWatchChannel { + channel_id: Uuid::new_v4(), + resource_id: "resource-expired".to_string(), + expires_at: (Utc::now() - Duration::hours(1)).trunc_subsecs(6), + }; + repo.record_watch_channel(key, lease_token, account_id, calendar_id, channel) + .await + .unwrap(); + + assert!( + !repo + .list_active_watch_channels() + .await + .unwrap() + .iter() + .any(|entry| entry.calendar_id == calendar_id), + "an already-lapsed channel is not worth a stop call" + ); +} diff --git a/crates/calendar_watch_relay/Cargo.toml b/crates/calendar_watch_relay/Cargo.toml new file mode 100644 index 00000000000..c4bf5ce94d3 --- /dev/null +++ b/crates/calendar_watch_relay/Cargo.toml @@ -0,0 +1,14 @@ +[package] +edition = "2024" +name = "calendar_watch_relay" +publish = false +version = "0.1.0" + +[dependencies] +macro_env_var = { path = "../macro_env_var" } +serde = { workspace = true, features = ["derive"] } +sha2 = { workspace = true } +workspace-hack = { version = "0.1", path = "../workspace-hack" } + +[dev-dependencies] +serde_json = { workspace = true } diff --git a/crates/calendar_watch_relay/src/lib.rs b/crates/calendar_watch_relay/src/lib.rs new file mode 100644 index 00000000000..3e13df000e5 --- /dev/null +++ b/crates/calendar_watch_relay/src/lib.rs @@ -0,0 +1,106 @@ +#![deny(missing_docs)] +//! Wire types and helpers shared by the calendar watch relay. +//! +//! Google requires an `events.watch` channel's address to be public HTTPS on +//! a domain verified in the Cloud project owning the OAuth client, which a +//! laptop can never satisfy. The relay closes that gap for local stacks: +//! their channels open against the dev-deployed +//! `calendar-event-local-tunnel` service's public address with a +//! per-instance token, and the stack's pubsub workers connect OUT to that +//! service and subscribe (SSE) to deliveries addressed to their token: +//! +//! ```text +//! Google ──POST──▶ calendar-event-local-tunnel (dev) +//! │ route by x-goog-channel-token +//! ▼ SSE (outbound connection from the laptop) +//! local stack re-injects the ping into its own +//! `handle_watch_notification` flow +//! ``` +//! +//! This crate carries what both ends must agree on — the wire model, the +//! subscriber's SSE parsing, the secret comparison, and the env-var readers +//! — while the tunnel service owns delivery fan-out and the email service +//! owns the subscriber loop. + +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +/// One relayed push notification: the meaningful subset of Google's +/// `x-goog-*` headers, in the wire shape shared by the tunnel's SSE stream +/// and its subscribers. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RelayedWatchNotification { + /// `x-goog-resource-state`: `sync`, `exists`, or `not_exists`. + pub state: String, + /// `x-goog-channel-id`. + pub channel_id: String, + /// `x-goog-resource-id`. + pub resource_id: String, +} + +/// Subscriber-side relay configuration, present only on stacks that consume +/// relayed deliveries (local). +pub struct WatchRelaySubscriberConfig { + /// Base URL of the tunnel deployment, e.g. + /// `https://calendar-event-local-tunnel-dev.macro.com`. + pub url: String, + /// Shared secret presented when subscribing. + pub secret: String, +} + +/// Read one environment variable, treating blank as unset like every other +/// watch variable. +pub fn read_env(name: &'static str) -> Option { + macro_env_var::maybe_read_env(name) + .map(|value| value.trim().to_owned()) + .filter(|value| !value.is_empty()) +} + +/// Read the subscriber-side configuration. +pub fn watch_relay_subscriber_config() -> Option { + let url = read_env("CALENDAR_WATCH_RELAY_URL")?; + let secret = read_env("CALENDAR_WATCH_RELAY_SECRET")?; + Some(WatchRelaySubscriberConfig { + url: url.trim_end_matches('/').to_owned(), + secret, + }) +} + +/// Compare two secrets without early exit on the first differing byte. +pub fn secrets_match(presented: &str, expected: &str) -> bool { + Sha256::digest(presented.as_bytes()) == Sha256::digest(expected.as_bytes()) +} + +/// Incremental parser extracting `data:` payloads from an SSE byte stream. +/// Comment lines (keep-alives) and other fields are ignored; multi-line data +/// is joined with `\n` per the SSE specification. +#[derive(Default)] +pub struct SseDataParser { + buffer: String, + data_lines: Vec, +} + +impl SseDataParser { + /// Feed one chunk, returning every event payload it completed. + pub fn push(&mut self, chunk: &str) -> Vec { + self.buffer.push_str(chunk); + let mut completed = Vec::new(); + while let Some(newline) = self.buffer.find('\n') { + let line: String = self.buffer.drain(..=newline).collect(); + let line = line.trim_end_matches(['\n', '\r']); + if line.is_empty() { + if !self.data_lines.is_empty() { + completed.push(self.data_lines.join("\n")); + self.data_lines.clear(); + } + } else if let Some(data) = line.strip_prefix("data:") { + self.data_lines + .push(data.strip_prefix(' ').unwrap_or(data).to_owned()); + } + } + completed + } +} + +#[cfg(test)] +mod test; diff --git a/crates/calendar_watch_relay/src/test.rs b/crates/calendar_watch_relay/src/test.rs new file mode 100644 index 00000000000..5af00ad101a --- /dev/null +++ b/crates/calendar_watch_relay/src/test.rs @@ -0,0 +1,50 @@ +use super::*; + +#[test] +fn sse_parser_extracts_single_event() { + let mut parser = SseDataParser::default(); + let events = parser.push("data: {\"a\":1}\n\n"); + assert_eq!(events, vec!["{\"a\":1}".to_owned()]); +} + +#[test] +fn sse_parser_handles_chunks_split_mid_line() { + let mut parser = SseDataParser::default(); + assert!(parser.push("data: {\"a\"").is_empty()); + assert!(parser.push(":1}\n").is_empty()); + let events = parser.push("\n"); + assert_eq!(events, vec!["{\"a\":1}".to_owned()]); +} + +#[test] +fn sse_parser_ignores_comments_and_other_fields() { + let mut parser = SseDataParser::default(); + let events = parser.push(": keep-alive\n\nevent: message\ndata: x\nid: 7\n\n"); + assert_eq!(events, vec!["x".to_owned()]); +} + +#[test] +fn sse_parser_joins_multi_line_data_and_handles_crlf() { + let mut parser = SseDataParser::default(); + let events = parser.push("data: one\r\ndata: two\r\n\r\ndata:three\n\n"); + assert_eq!(events, vec!["one\ntwo".to_owned(), "three".to_owned()]); +} + +#[test] +fn secrets_match_requires_equality() { + assert!(secrets_match("s3cret", "s3cret")); + assert!(!secrets_match("s3cret", "s3cret ")); + assert!(!secrets_match("", "s3cret")); +} + +#[test] +fn relayed_notification_wire_round_trip() { + let notification = RelayedWatchNotification { + state: "exists".to_owned(), + channel_id: "chan".to_owned(), + resource_id: "res".to_owned(), + }; + let encoded = serde_json::to_string(¬ification).unwrap(); + let decoded: RelayedWatchNotification = serde_json::from_str(&encoded).unwrap(); + assert_eq!(decoded, notification); +} diff --git a/crates/workspace-hack/Cargo.toml b/crates/workspace-hack/Cargo.toml index e0106708154..be0be18adb3 100644 --- a/crates/workspace-hack/Cargo.toml +++ b/crates/workspace-hack/Cargo.toml @@ -124,7 +124,7 @@ thiserror = { version = "2" } time = { version = "0.3", features = ["formatting", "macros", "parsing"] } tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "parking_lot", "process", "rt-multi-thread", "signal", "test-util"] } tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "tls12"] } -tokio-stream = { version = "0.1", features = ["fs"] } +tokio-stream = { version = "0.1", features = ["fs", "sync"] } tokio-util = { version = "0.7", features = ["codec", "compat", "io", "rt"] } tower = { version = "0.5", default-features = false, features = ["balance", "buffer", "limit", "load-shed", "log"] } tower-http = { version = "0.6", features = ["compression-gzip", "cors", "fs", "limit", "request-id", "timeout", "trace", "util"] } @@ -256,7 +256,7 @@ thiserror = { version = "2" } time = { version = "0.3", features = ["formatting", "macros", "parsing"] } tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "parking_lot", "process", "rt-multi-thread", "signal", "test-util"] } tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "tls12"] } -tokio-stream = { version = "0.1", features = ["fs"] } +tokio-stream = { version = "0.1", features = ["fs", "sync"] } tokio-util = { version = "0.7", features = ["codec", "compat", "io", "rt"] } tower = { version = "0.5", default-features = false, features = ["balance", "buffer", "limit", "load-shed", "log"] } tower-http = { version = "0.6", features = ["compression-gzip", "cors", "fs", "limit", "request-id", "timeout", "trace", "util"] } diff --git a/docs/RUNNING_LOCALLY.md b/docs/RUNNING_LOCALLY.md index 10181923289..896d276f01c 100644 --- a/docs/RUNNING_LOCALLY.md +++ b/docs/RUNNING_LOCALLY.md @@ -314,3 +314,55 @@ just reset_local --instance agent-a ``` For the default instance, omit `--instance`. + +## Real Google Calendar Push (`calendar-push`) + +By default a local stack only notices provider-side calendar edits on its +5-minute poll. Calendar push closes that gap to seconds by relaying Google's +`events.watch` webhook deliveries through the dev-only +`calendar-event-local-tunnel` service: Google requires a watch channel's +address to be public HTTPS on a domain verified in the Cloud project owning +the OAuth client, which a laptop can never satisfy, so local channels open +against `calendar-event-local-tunnel-dev.macro.com` with a per-instance +token, and the local pubsub workers subscribe OUT to that service (SSE) for +deliveries addressed to their token. No tunnel daemon, no DNS, no inbound +connectivity to your machine. + +```bash +just calendar-push enable --instance agent-a # or omit --instance +just run_local --instance agent-a # (re)start to apply +``` + +`enable` writes `infra/local/generated//calendar-push.env`, which +every env resolve overlays automatically. It requires +`CALENDAR_WATCH_RELAY_SECRET`, pulled with the rest of the Doppler +`local/lcl_personal` config (or supplied via `--env-file`); the resolve warns +if it is missing. + +To verify: connect a Google account, then edit an event in the Google +Calendar UI — the local projection should update within seconds. The pubsub +workers log `subscribing to relayed calendar watch notifications` on start +and `relayed watch channel handshake received` when a channel opens. + +Turning it off: + +```bash +just calendar-push disable --instance agent-a +``` + +A running stack keeps push until it restarts. On graceful shutdown (stop, +destroy, restart) the stack calls `channels.stop` at Google for every open +channel; anything that slips through lapses at its natural expiry, and the +tunnel drops those strays centrally, so they never reach your machine. + +The no-Google fake path still works for plumbing checks: stamp +`watch_channel_id`/`watch_resource_id` on a `calendars` row, then POST to the +local webhook directly: + +```bash +curl -i -X POST http://localhost:/calendar/notifications \ + -H "x-goog-channel-token: " \ + -H "x-goog-channel-id: " \ + -H "x-goog-resource-id: " \ + -H "x-goog-resource-state: exists" +``` diff --git a/infra/stacks/calendar-event-local-tunnel/Pulumi.dev.yaml b/infra/stacks/calendar-event-local-tunnel/Pulumi.dev.yaml new file mode 100644 index 00000000000..9cfd7a1eb50 --- /dev/null +++ b/infra/stacks/calendar-event-local-tunnel/Pulumi.dev.yaml @@ -0,0 +1,3 @@ +config: + aws-native:region: us-east-1 + aws:region: us-east-1 diff --git a/infra/stacks/calendar-event-local-tunnel/Pulumi.prod.yaml b/infra/stacks/calendar-event-local-tunnel/Pulumi.prod.yaml new file mode 100644 index 00000000000..9cfd7a1eb50 --- /dev/null +++ b/infra/stacks/calendar-event-local-tunnel/Pulumi.prod.yaml @@ -0,0 +1,3 @@ +config: + aws-native:region: us-east-1 + aws:region: us-east-1 diff --git a/infra/stacks/calendar-event-local-tunnel/Pulumi.yaml b/infra/stacks/calendar-event-local-tunnel/Pulumi.yaml new file mode 100644 index 00000000000..aa2f6bf1240 --- /dev/null +++ b/infra/stacks/calendar-event-local-tunnel/Pulumi.yaml @@ -0,0 +1,3 @@ +name: calendar-event-local-tunnel +runtime: nodejs +description: Dev-only tunnel relaying Google Calendar push notifications to local stacks diff --git a/infra/stacks/calendar-event-local-tunnel/calendar-event-local-tunnel.ts b/infra/stacks/calendar-event-local-tunnel/calendar-event-local-tunnel.ts new file mode 100644 index 00000000000..a92e63042ca --- /dev/null +++ b/infra/stacks/calendar-event-local-tunnel/calendar-event-local-tunnel.ts @@ -0,0 +1,430 @@ +import * as aws from '@pulumi/aws'; +import * as awsx from '@pulumi/awsx'; +import * as pulumi from '@pulumi/pulumi'; +import { + DATADOG_API_KEY, + DEFAULT_CONTINUE_BEFORE_STEADY_STATE, + EcsDeploymentFailureAlarm, + datadogAgentContainer, + fargateLogRouterSidecarContainer, + serviceLoadBalancer, +} from '../../packages/resources'; +import { EcrImage } from '../../packages/service'; +import { + BASE_DOMAIN, + CLOUD_TRAIL_SNS_TOPIC_ARN, + DopplerEcsEnvironment, + stack, +} from '../../packages/shared'; + +const BASE_NAME = pulumi.getProject(); +const REPO_ROOT = '../../..'; + +export const SERVICE_DOMAIN_NAME = `calendar-event-local-tunnel${ + stack === 'prod' ? '' : `-${stack}` +}.${BASE_DOMAIN}`; + +type CreateCalendarEventLocalTunnelArgs = { + cloudStorageClusterName: pulumi.Output | string; + ecsClusterArn: pulumi.Output | string; + vpc: { + vpcId: pulumi.Output | string; + publicSubnetIds: pulumi.Output | string[]; + privateSubnetIds: pulumi.Output | string[]; + }; + platform: { family: string; architecture: 'amd64' | 'arm64' }; + serviceContainerPort: number; + isPrivate?: boolean; + containerEnvVars?: { name: string; value: pulumi.Output | string }[]; + healthCheckPath: string; + tags: { [key: string]: string }; +}; + +export class CalendarEventLocalTunnel extends pulumi.ComponentResource { + public ecr: awsx.ecr.Repository; + public serviceAlbSg: aws.ec2.SecurityGroup; + public serviceSg: aws.ec2.SecurityGroup; + public targetGroup: aws.lb.TargetGroup; + public lb: aws.lb.LoadBalancer; + public listener: aws.lb.Listener; + public service: awsx.ecs.FargateService; + public domain: string; + public cloudStorageClusterName: pulumi.Output | string; + public tags: { [key: string]: string }; + public role: aws.iam.Role; + + constructor( + name: string, + { + ecsClusterArn, + vpc, + platform, + serviceContainerPort, + healthCheckPath, + isPrivate, + containerEnvVars, + cloudStorageClusterName, + tags, + }: CreateCalendarEventLocalTunnelArgs, + opts?: pulumi.ComponentResourceOptions + ) { + super('my:components:CalendarEventLocalTunnel', name, {}, opts); + this.tags = tags; + + this.cloudStorageClusterName = cloudStorageClusterName; + + // ecr image + const image = new EcrImage( + `${BASE_NAME}-ecr-image-${stack}`, + { + repositoryId: `${BASE_NAME}-ecr-${stack}`, + repositoryName: `${BASE_NAME}-${stack}`, + imageId: `${BASE_NAME}-image-${stack}`, + imagePath: REPO_ROOT, + dockerfile: 'docker/Dockerfile', + platform, + buildArgs: { + SERVICE_NAME: 'calendar_event_local_tunnel', + }, + tags: this.tags, + }, + { parent: this } + ); + this.ecr = image.ecr; + + // sg + const sg = this.initializeSecurityGroups({ + vpcId: vpc.vpcId, + serviceContainerPort, + }); + this.serviceAlbSg = sg.serviceAlbSg; + this.serviceSg = sg.serviceSg; + + // lb — long idle timeout: subscribers hold SSE connections open and the + // service keep-alives every 15 seconds. + const { targetGroup, lb, listener } = serviceLoadBalancer(this, { + serviceName: BASE_NAME, + serviceContainerPort, + healthCheckPath, + vpc, + albSecurityGroupId: this.serviceAlbSg.id, + isPrivate, + tags, + idleTimeout: 3600, + }); + this.targetGroup = targetGroup; + this.lb = lb; + this.listener = listener; + + this.role = new aws.iam.Role( + `${BASE_NAME}-role`, + { + name: `${BASE_NAME}-role-${stack}`, + assumeRolePolicy: { + Version: '2012-10-17', + Statement: [ + { + Action: 'sts:AssumeRole', + Principal: { + Service: 'ecs-tasks.amazonaws.com', + }, + Effect: 'Allow', + Sid: '', + }, + ], + }, + managedPolicyArns: [], + tags: this.tags, + }, + { parent: this } + ); + + const dopplerEcsEnvironment = new DopplerEcsEnvironment( + BASE_NAME, + { tags: this.tags }, + { parent: this } + ); + + // service — pinned to a single task with no autoscaling: subscriber + // fan-out is an in-memory map, so a webhook landing on one task while + // the matching subscriber is connected to another would drop deliveries. + const service = new awsx.ecs.FargateService( + `${BASE_NAME}`, + { + tags, + cluster: ecsClusterArn, + networkConfiguration: { + subnets: vpc.privateSubnetIds, + securityGroups: [this.serviceSg.id], + }, + continueBeforeSteadyState: DEFAULT_CONTINUE_BEFORE_STEADY_STATE, + deploymentCircuitBreaker: { + enable: true, + rollback: true, + }, + taskDefinitionArgs: { + taskRole: { + roleArn: this.role.arn, + }, + executionRole: { + roleArn: dopplerEcsEnvironment.executionRole.arn, + }, + + containers: { + log_router: fargateLogRouterSidecarContainer, + datadog_agent: datadogAgentContainer, + service: { + name: BASE_NAME, + image: image.image.imageUri, + stopTimeout: 10, // 10 seconds to force kill the task + cpu: 256, + memory: 512, + environment: containerEnvVars, + secrets: [...dopplerEcsEnvironment.containerSecrets], + logConfiguration: { + logDriver: 'awsfirelens', + options: { + Name: 'datadog', + Host: 'http-intake.logs.us5.datadoghq.com', + apikey: DATADOG_API_KEY, + dd_service: 'calendar-event-local-tunnel', + dd_source: 'fargate', + dd_tags: `project:calendar-event-local-tunnel, env:${stack}`, + provider: 'ecs', + }, + }, + portMappings: [ + { + appProtocol: 'http', + name: `${BASE_NAME}-tcp-${stack}`, + hostPort: serviceContainerPort, + containerPort: serviceContainerPort, + targetGroup, + }, + ], + }, + }, + runtimePlatform: { + operatingSystemFamily: `${platform.family.toUpperCase()}`, + cpuArchitecture: `${ + platform.architecture === 'amd64' + ? 'X86_64' + : platform.architecture.toUpperCase() + }`, + }, + }, + desiredCount: 1, + }, + { + parent: this, + } + ); + + this.service = service; + + this.setupServiceAlarms(); + + // domain record + const zone = aws.route53.getZoneOutput({ name: BASE_DOMAIN }); + + new aws.route53.Record( + `${BASE_NAME}-domain-record`, + { + name: `${SERVICE_DOMAIN_NAME}`, + type: 'A', + zoneId: zone.zoneId, + aliases: [ + { + evaluateTargetHealth: false, + name: this.lb.dnsName, + zoneId: this.lb.zoneId, + }, + ], + }, + { parent: this } + ); + + this.domain = `https://${SERVICE_DOMAIN_NAME}`; + } + + initializeSecurityGroups({ + vpcId, + serviceContainerPort, + }: { + vpcId: pulumi.Output | string; + serviceContainerPort: number; + }) { + const serviceAlbSg = new aws.ec2.SecurityGroup( + `${BASE_NAME}-alb-sg-${stack}`, + { + name: `${BASE_NAME}-alb-sg-${stack}`, + description: `${BASE_NAME} application load balancer security group`, + vpcId, + tags: this.tags, + }, + { parent: this } + ); + + const serviceSg = new aws.ec2.SecurityGroup( + `${BASE_NAME}-sg-${stack}`, + { + name: `${BASE_NAME}-sg-${stack}`, + vpcId, + description: `${BASE_NAME} security group that is attached directly to the service`, + tags: this.tags, + }, + { parent: this } + ); + + new aws.vpc.SecurityGroupIngressRule( + `${BASE_NAME}-alb-in`, + { + securityGroupId: serviceSg.id, + description: 'Allow inbound traffic from the services ALB', + referencedSecurityGroupId: serviceAlbSg.id, + fromPort: serviceContainerPort, + toPort: serviceContainerPort, + ipProtocol: 'tcp', + tags: this.tags, + }, + { parent: this } + ); + + new aws.vpc.SecurityGroupEgressRule( + `${BASE_NAME}-all-out`, + { + securityGroupId: serviceSg.id, + description: 'Allow all outbound', + cidrIpv4: '0.0.0.0/0', + ipProtocol: '-1', + tags: this.tags, + }, + { parent: this } + ); + + // ALB SG rules + new aws.vpc.SecurityGroupIngressRule( + `${BASE_NAME}-http`, + { + securityGroupId: serviceAlbSg.id, + description: 'Allow inbound HTTP traffic', + cidrIpv4: '0.0.0.0/0', + fromPort: 80, + ipProtocol: 'tcp', + toPort: 80, + tags: this.tags, + }, + { parent: this } + ); + + new aws.vpc.SecurityGroupIngressRule( + `${BASE_NAME}-https`, + { + securityGroupId: serviceAlbSg.id, + description: 'Allow inbound HTTPS traffic', + cidrIpv4: '0.0.0.0/0', + fromPort: 443, + ipProtocol: 'tcp', + toPort: 443, + tags: this.tags, + }, + { parent: this } + ); + + new aws.vpc.SecurityGroupEgressRule( + `${BASE_NAME}-out-service`, + { + description: 'Allow traffic to the service security group', + securityGroupId: serviceAlbSg.id, + referencedSecurityGroupId: serviceSg.id, + fromPort: serviceContainerPort, + ipProtocol: 'tcp', + toPort: serviceContainerPort, + tags: this.tags, + }, + { parent: this } + ); + + return { serviceAlbSg, serviceSg }; + } + + setupServiceAlarms() { + new EcsDeploymentFailureAlarm( + `${BASE_NAME}-deployment-failure-alarm`, + { + serviceName: BASE_NAME, + serviceArn: this.service.service.arn, + tags: this.tags, + }, + { parent: this } + ); + + new aws.cloudwatch.MetricAlarm( + `${BASE_NAME}-high-cpu-alarm`, + { + name: `${BASE_NAME}-high-cpu-alarm-${stack}`, + metricName: 'CPUUtilization', + namespace: 'AWS/ECS', + statistic: 'Average', + period: 180, + evaluationPeriods: 1, + threshold: 80, + comparisonOperator: 'GreaterThanThreshold', + dimensions: { + ClusterName: this.cloudStorageClusterName, + ServiceName: this.service.service.name, + }, + alarmDescription: `High CPU usage alarm for ${BASE_NAME} service.`, + actionsEnabled: true, + alarmActions: [CLOUD_TRAIL_SNS_TOPIC_ARN], + tags: this.tags, + }, + { parent: this } + ); + + new aws.cloudwatch.MetricAlarm( + `${BASE_NAME}-high-mem-alarm`, + { + name: `${BASE_NAME}-high-mem-alarm-${stack}`, + metricName: 'MemoryUtilization', + namespace: 'AWS/ECS', + statistic: 'Average', + period: 180, + evaluationPeriods: 1, + threshold: 80, + comparisonOperator: 'GreaterThanThreshold', + dimensions: { + ClusterName: this.cloudStorageClusterName, + ServiceName: this.service.service.name, + }, + alarmDescription: `High Memory usage alarm for ${BASE_NAME} service.`, + actionsEnabled: true, + alarmActions: [CLOUD_TRAIL_SNS_TOPIC_ARN], + tags: this.tags, + }, + { parent: this } + ); + + new aws.cloudwatch.MetricAlarm( + `${BASE_NAME}-http-5xx-alarm`, + { + name: `${BASE_NAME}-http-5xx-${stack}`, + metricName: 'HTTPCode_ELB_5XX_Count', + namespace: 'AWS/ApplicationELB', + statistic: 'Sum', + period: 180, + evaluationPeriods: 1, + threshold: 25, + comparisonOperator: 'GreaterThanOrEqualToThreshold', + dimensions: { + LoadBalancer: this.lb.arn, + }, + alarmDescription: `High HTTP 5XX count alarm for ${BASE_NAME} Load Balancer.`, + actionsEnabled: true, + alarmActions: [CLOUD_TRAIL_SNS_TOPIC_ARN], + tags: this.tags, + }, + { parent: this } + ); + } +} diff --git a/infra/stacks/calendar-event-local-tunnel/index.ts b/infra/stacks/calendar-event-local-tunnel/index.ts new file mode 100644 index 00000000000..a38380cc52d --- /dev/null +++ b/infra/stacks/calendar-event-local-tunnel/index.ts @@ -0,0 +1,68 @@ +import * as pulumi from '@pulumi/pulumi'; +import { stack } from '../../packages/shared'; +import { get_coparse_api_vpc } from '../../packages/vpc'; +import { CalendarEventLocalTunnel } from './calendar-event-local-tunnel'; + +const tags = { + environment: stack, + tech_lead: 'gabriel', + project: 'calendar-event-local-tunnel', +}; + +// Dev-only: the tunnel exists so locally running stacks can receive real +// Google Calendar push notifications; there is nothing for it to relay in +// prod. The prod stack stays an empty no-op so environment-wide deploy runs +// succeed. +let tunnel: CalendarEventLocalTunnel | undefined; + +if (stack === 'dev') { + const coparse_api_vpc = get_coparse_api_vpc(); + + const cloudStorageStack = new pulumi.StackReference('cloud-storage-stack', { + name: `macro-inc/document-storage/${stack}`, + }); + + const cloudStorageClusterArn: pulumi.Output = cloudStorageStack + .getOutput('cloudStorageClusterArn') + .apply((arn) => arn as string); + + const cloudStorageClusterName: pulumi.Output = cloudStorageStack + .getOutput('cloudStorageClusterName') + .apply((arn) => arn as string); + + tunnel = new CalendarEventLocalTunnel( + `calendar-event-local-tunnel-${stack}`, + { + ecsClusterArn: cloudStorageClusterArn, + cloudStorageClusterName: cloudStorageClusterName, + vpc: coparse_api_vpc, + platform: { + family: 'linux', + architecture: 'amd64', + }, + serviceContainerPort: 8080, + healthCheckPath: '/health', + containerEnvVars: [ + { + name: 'ENVIRONMENT', + value: stack, + }, + // OpenTelemetry / Datadog tracing configuration + { + name: 'DD_SERVICE', + value: 'calendar-event-local-tunnel', + }, + { + name: 'DD_ENV', + value: stack, + }, + ], + isPrivate: false, + tags, + } + ); +} + +export const calendarEventLocalTunnelUrl = tunnel + ? pulumi.interpolate`${tunnel.domain}` + : undefined; diff --git a/infra/stacks/calendar-event-local-tunnel/package.json b/infra/stacks/calendar-event-local-tunnel/package.json new file mode 100644 index 00000000000..cb04765a595 --- /dev/null +++ b/infra/stacks/calendar-event-local-tunnel/package.json @@ -0,0 +1,6 @@ +{ + "name": "calendar-event-local-tunnel-stack", + "version": "0.0.0", + "private": true, + "license": "MIT" +} diff --git a/infra/stacks/calendar-event-local-tunnel/tsconfig.json b/infra/stacks/calendar-event-local-tunnel/tsconfig.json new file mode 100644 index 00000000000..6f83eb665f8 --- /dev/null +++ b/infra/stacks/calendar-event-local-tunnel/tsconfig.json @@ -0,0 +1,3 @@ +{ + "extends": "../../tsconfig.json", +} diff --git a/infra/stacks/doppler-projects/index.ts b/infra/stacks/doppler-projects/index.ts index a8c64f61346..82b7e3ff036 100644 --- a/infra/stacks/doppler-projects/index.ts +++ b/infra/stacks/doppler-projects/index.ts @@ -15,6 +15,7 @@ const SERVICE_NAMES = [ 'mcp-server', 'email-service', 'document-cognition', + 'calendar-event-local-tunnel', ]; for (const service_name of SERVICE_NAMES) { diff --git a/nix/cloud-storage.nix b/nix/cloud-storage.nix index 34cc488d4e9..12d2576b991 100644 --- a/nix/cloud-storage.nix +++ b/nix/cloud-storage.nix @@ -347,6 +347,11 @@ packageName = "authentication_service"; binaries = [ "authentication_service" ]; } + { + serviceName = "calendar-event-local-tunnel"; + packageName = "calendar_event_local_tunnel"; + binaries = [ "calendar_event_local_tunnel" ]; + } { serviceName = "connection-gateway"; packageName = "connection_gateway"; diff --git a/services/calendar_event_local_tunnel/Cargo.toml b/services/calendar_event_local_tunnel/Cargo.toml new file mode 100644 index 00000000000..6383ce99a52 --- /dev/null +++ b/services/calendar_event_local_tunnel/Cargo.toml @@ -0,0 +1,34 @@ +[package] +edition = "2024" +name = "calendar_event_local_tunnel" +publish = false +version = "0.1.0" +default-run = "calendar_event_local_tunnel" + +[[bin]] +name = "calendar_event_local_tunnel" +path = "src/main.rs" + +[[bin]] +name = "calendar_event_local_tunnel_doppler_config" +path = "src/doppler_config.rs" + +[dependencies] +anyhow = { workspace = true } +axum = { workspace = true } +calendar_watch_relay = { path = "../../crates/calendar_watch_relay" } +doppler-config = { workspace = true } +futures = { workspace = true } +macro_config = { path = "../../crates/macro_config" } +macro_entrypoint = { path = "../../crates/macro_entrypoint" } +macro_env = { path = "../../crates/macro_env" } +serde = { workspace = true } +tokio = { workspace = true } +tokio-stream = { workspace = true, features = ["sync"] } +tower = { workspace = true } +tower-http = { workspace = true, features = ["trace"] } +tracing = { workspace = true } +workspace-hack = { version = "0.1", path = "../../crates/workspace-hack" } + +[dev-dependencies] +serde_json = { workspace = true } diff --git a/services/calendar_event_local_tunnel/src/api.rs b/services/calendar_event_local_tunnel/src/api.rs new file mode 100644 index 00000000000..6f7d97bd5ea --- /dev/null +++ b/services/calendar_event_local_tunnel/src/api.rs @@ -0,0 +1,106 @@ +//! HTTP surface: Google's webhook in, subscriber SSE out. + +use std::sync::Arc; +use std::time::Duration; + +use axum::Router; +use axum::extract::State; +use axum::http::{HeaderMap, StatusCode}; +use axum::response::sse::{Event, KeepAlive, Sse}; +use axum::response::{IntoResponse, Response}; +use axum::routing::{get, post}; +use calendar_watch_relay::{RelayedWatchNotification, secrets_match}; +use futures::StreamExt; +use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::wrappers::errors::BroadcastStreamRecvError; + +use crate::registry::RelayRegistry; + +/// Shared handler state. +#[derive(Clone)] +pub struct ApiContext { + /// Live subscriptions by channel token. + pub registry: RelayRegistry, + /// Shared secret subscribers must present. + pub secret: Arc, +} + +/// Build the service router. +pub fn router(state: ApiContext) -> Router { + Router::new() + .route("/health", get(health)) + .route("/calendar/notifications", post(notifications)) + .route("/calendar/relay/subscribe", get(subscribe)) + .with_state(state) +} + +async fn health() -> StatusCode { + StatusCode::OK +} + +fn header<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> { + headers.get(name).and_then(|value| value.to_str().ok()) +} + +/// Google's webhook. Deliveries are routed purely by channel token; the +/// token was minted by the local stack that opened the channel, so a +/// delivery nobody subscribes to (a stray from a torn-down stack, or a +/// probe with an invented token) is acknowledged and dropped. +#[tracing::instrument(skip_all)] +async fn notifications(State(ctx): State, headers: HeaderMap) -> StatusCode { + let Some(token) = header(&headers, "x-goog-channel-token") else { + return StatusCode::FORBIDDEN; + }; + let (Some(state), Some(channel_id), Some(resource_id)) = ( + header(&headers, "x-goog-resource-state"), + header(&headers, "x-goog-channel-id"), + header(&headers, "x-goog-resource-id"), + ) else { + return StatusCode::BAD_REQUEST; + }; + let notification = RelayedWatchNotification { + state: state.to_owned(), + channel_id: channel_id.to_owned(), + resource_id: resource_id.to_owned(), + }; + let delivered = ctx.registry.publish(token, notification); + if delivered == 0 { + tracing::debug!(channel_id, "calendar notification matched no subscriber"); + } + StatusCode::OK +} + +/// Stream relayed notifications for one channel token over SSE. +/// +/// Subscribers authenticate with the shared relay secret; the stream then +/// carries only deliveries addressed to the presented token, so one local +/// stack can never observe another's notifications. +#[tracing::instrument(skip_all)] +async fn subscribe(State(ctx): State, headers: HeaderMap) -> Response { + let Some(secret) = header(&headers, "x-relay-secret") else { + return StatusCode::FORBIDDEN.into_response(); + }; + if !secrets_match(secret, &ctx.secret) { + return StatusCode::FORBIDDEN.into_response(); + } + let Some(token) = header(&headers, "x-relay-token") else { + return StatusCode::BAD_REQUEST.into_response(); + }; + let stream = + BroadcastStream::new(ctx.registry.subscribe(token)).filter_map(|delivery| async move { + match delivery { + Ok(notification) => Some(Event::default().json_data(¬ification)), + // Dropped deliveries degrade to the subscriber's poll backstop. + Err(BroadcastStreamRecvError::Lagged(dropped)) => { + tracing::warn!(dropped, "relay subscriber lagged; deliveries dropped"); + None + } + } + }); + Sse::new(stream) + .keep_alive(KeepAlive::new().interval(Duration::from_secs(15))) + .into_response() +} + +#[cfg(test)] +mod test; diff --git a/services/calendar_event_local_tunnel/src/api/test.rs b/services/calendar_event_local_tunnel/src/api/test.rs new file mode 100644 index 00000000000..2c45658b4e1 --- /dev/null +++ b/services/calendar_event_local_tunnel/src/api/test.rs @@ -0,0 +1,130 @@ +use super::*; + +use axum::body::Body; +use axum::http::Request; +use tower::ServiceExt; + +fn app() -> Router { + router(ApiContext { + registry: RelayRegistry::default(), + secret: Arc::new("s3cret".to_owned()), + }) +} + +fn notification_request(token: Option<&str>, complete: bool) -> Request { + let mut builder = Request::builder() + .method("POST") + .uri("/calendar/notifications"); + if let Some(token) = token { + builder = builder.header("x-goog-channel-token", token); + } + if complete { + builder = builder + .header("x-goog-resource-state", "exists") + .header("x-goog-channel-id", "chan-1") + .header("x-goog-resource-id", "res-1"); + } + builder.body(Body::empty()).unwrap() +} + +#[tokio::test] +async fn health_answers_ok() { + let response = app() + .oneshot(Request::get("/health").body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); +} + +#[tokio::test] +async fn notifications_require_a_token_and_the_goog_headers() { + let app = app(); + let missing_token = app + .clone() + .oneshot(notification_request(None, true)) + .await + .unwrap(); + assert_eq!(missing_token.status(), StatusCode::FORBIDDEN); + + let missing_headers = app + .clone() + .oneshot(notification_request(Some("t"), false)) + .await + .unwrap(); + assert_eq!(missing_headers.status(), StatusCode::BAD_REQUEST); + + let stray = app + .oneshot(notification_request(Some("nobody-listening"), true)) + .await + .unwrap(); + assert_eq!(stray.status(), StatusCode::OK, "strays are acknowledged"); +} + +#[tokio::test] +async fn subscribe_requires_the_shared_secret_and_a_token() { + let app = app(); + let wrong_secret = app + .clone() + .oneshot( + Request::get("/calendar/relay/subscribe") + .header("x-relay-secret", "wrong") + .header("x-relay-token", "t") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(wrong_secret.status(), StatusCode::FORBIDDEN); + + let missing_token = app + .oneshot( + Request::get("/calendar/relay/subscribe") + .header("x-relay-secret", "s3cret") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(missing_token.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn a_delivery_reaches_a_live_subscriber_as_an_sse_event() { + let state = ApiContext { + registry: RelayRegistry::default(), + secret: Arc::new("s3cret".to_owned()), + }; + let subscription = router(state.clone()) + .oneshot( + Request::get("/calendar/relay/subscribe") + .header("x-relay-secret", "s3cret") + .header("x-relay-token", "token-e2e") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(subscription.status(), StatusCode::OK); + + let delivery = router(state) + .oneshot(notification_request(Some("token-e2e"), true)) + .await + .unwrap(); + assert_eq!(delivery.status(), StatusCode::OK); + + let frame = tokio::time::timeout( + Duration::from_secs(5), + subscription.into_body().into_data_stream().next(), + ) + .await + .expect("a frame arrives promptly") + .expect("the stream is open") + .unwrap(); + let text = String::from_utf8(frame.to_vec()).unwrap(); + assert!(text.starts_with("data:"), "unexpected frame: {text}"); + let payload: RelayedWatchNotification = + serde_json::from_str(text.trim_start_matches("data:").trim()).unwrap(); + assert_eq!(payload.channel_id, "chan-1"); + assert_eq!(payload.resource_id, "res-1"); + assert_eq!(payload.state, "exists"); +} diff --git a/services/calendar_event_local_tunnel/src/config.rs b/services/calendar_event_local_tunnel/src/config.rs new file mode 100644 index 00000000000..bc417e8b762 --- /dev/null +++ b/services/calendar_event_local_tunnel/src/config.rs @@ -0,0 +1,27 @@ +//! Service configuration. + +use anyhow::Context; +pub use macro_env::Environment; + +/// Environment-derived configuration. +#[derive(macro_config::MacroConfig)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub struct Config { + /// The port to listen on. + #[macro_config_default(8080)] + pub port: usize, + /// The environment we are in. + #[macro_config_default(Environment::new_or_prod())] + pub environment: Environment, + /// Shared secret subscribers must present. Required: a tunnel that + /// cannot authenticate subscribers must not start. + pub calendar_watch_relay_secret: String, +} + +impl Config { + /// Load the configuration from the environment. + pub fn from_env() -> anyhow::Result { + macro_config::ConfigLoader::load::() + .context("failed to load calendar event local tunnel config") + } +} diff --git a/services/calendar_event_local_tunnel/src/doppler_config.rs b/services/calendar_event_local_tunnel/src/doppler_config.rs new file mode 100644 index 00000000000..a422696ba13 --- /dev/null +++ b/services/calendar_event_local_tunnel/src/doppler_config.rs @@ -0,0 +1,21 @@ +#![allow(unused)] + +use macro_env::Environment; + +mod config; + +const DOPPLER_PROJECT: &str = "calendar-event-local-tunnel"; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + // Dev-only service: there is no prd config to validate. + let dev = doppler_config::DopplerConfig::builder() + .token_from_env("DOPPLER_TOKEN") + .config(Environment::Develop.to_doppler_slug()) + .project(DOPPLER_PROJECT) + .build() + .expect("able to grab doppler project"); + + dev.load::().await?; + Ok(()) +} diff --git a/services/calendar_event_local_tunnel/src/main.rs b/services/calendar_event_local_tunnel/src/main.rs new file mode 100644 index 00000000000..f773899c8d2 --- /dev/null +++ b/services/calendar_event_local_tunnel/src/main.rs @@ -0,0 +1,51 @@ +//! Dev-only tunnel relaying Google Calendar push notifications to local stacks. +//! +//! Local `run_local` stacks cannot receive Google's `events.watch` webhooks — +//! the callback address must be public HTTPS on a domain verified in the +//! Cloud project owning the OAuth client. This service is that address: +//! local channels open against `/calendar/notifications` here with a +//! per-instance token, and each stack's pubsub workers subscribe OUT to +//! `/calendar/relay/subscribe` (SSE) for deliveries addressed to their +//! token. Deliveries are content-free header triples; a token with no live +//! subscriber (a torn-down stack's strays) is dropped on the floor. +//! +//! The service is deliberately stateless and dependency-free: fan-out is an +//! in-memory map, so it deploys as a single dev task. A restart drops +//! subscriber connections; they reconnect with backoff, and every local +//! stack's 5-minute poll remains the freshness backstop throughout. + +mod api; +mod config; +mod registry; + +use std::sync::Arc; + +use anyhow::Context; +use macro_entrypoint::MacroEntrypoint; + +#[tokio::main] +#[tracing::instrument(err)] +async fn main() -> anyhow::Result<()> { + MacroEntrypoint::default().init(); + let config = config::Config::from_env().context("expected to be able to generate config")?; + if config.calendar_watch_relay_secret.trim().is_empty() { + anyhow::bail!("CALENDAR_WATCH_RELAY_SECRET must not be blank"); + } + let state = api::ApiContext { + registry: registry::RelayRegistry::default(), + secret: Arc::new(config.calendar_watch_relay_secret), + }; + let app = api::router(state).layer(tower_http::trace::TraceLayer::new_for_http()); + let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", config.port)) + .await + .with_context(|| format!("binding 0.0.0.0:{}", config.port))?; + tracing::info!( + environment = ?config.environment, + port = config.port, + "calendar-event-local-tunnel is up" + ); + axum::serve(listener, app.into_make_service()) + .with_graceful_shutdown(macro_entrypoint::shutdown_signal()) + .await + .context("error starting service") +} diff --git a/services/calendar_event_local_tunnel/src/registry.rs b/services/calendar_event_local_tunnel/src/registry.rs new file mode 100644 index 00000000000..abecb0e5c2a --- /dev/null +++ b/services/calendar_event_local_tunnel/src/registry.rs @@ -0,0 +1,54 @@ +//! In-memory fan-out from webhook deliveries to SSE subscribers. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use calendar_watch_relay::RelayedWatchNotification; +use tokio::sync::broadcast; + +/// Deliveries a slow subscriber may buffer before lagging. Lagged +/// notifications are dropped — the subscriber's poll remains the backstop. +const CHANNEL_CAPACITY: usize = 64; + +/// Live subscriptions keyed by channel token. +/// +/// Entries are reclaimed on the first publish after their last subscriber +/// disconnects; a token that subscribes and never receives a delivery keeps +/// one idle entry, which is bounded by the number of developers who ever +/// connected since the last deploy. +#[derive(Clone, Default)] +pub struct RelayRegistry { + inner: Arc>>>, +} + +impl RelayRegistry { + /// Deliver one notification to `token`'s subscribers, returning how many + /// received it. A token with no live subscriber is dropped on the floor — + /// that is exactly the stray case after a local stack is torn down. + pub fn publish(&self, token: &str, notification: RelayedWatchNotification) -> usize { + let mut inner = self.inner.lock().unwrap(); + let Some(sender) = inner.get(token) else { + return 0; + }; + match sender.send(notification) { + Ok(receivers) => receivers, + Err(_) => { + inner.remove(token); + 0 + } + } + } + + /// Open one subscription for `token`. + pub fn subscribe(&self, token: &str) -> broadcast::Receiver { + self.inner + .lock() + .unwrap() + .entry(token.to_owned()) + .or_insert_with(|| broadcast::channel(CHANNEL_CAPACITY).0) + .subscribe() + } +} + +#[cfg(test)] +mod test; diff --git a/services/calendar_event_local_tunnel/src/registry/test.rs b/services/calendar_event_local_tunnel/src/registry/test.rs new file mode 100644 index 00000000000..d5aec5c51f4 --- /dev/null +++ b/services/calendar_event_local_tunnel/src/registry/test.rs @@ -0,0 +1,50 @@ +use super::*; + +fn ping(channel_id: &str) -> RelayedWatchNotification { + RelayedWatchNotification { + state: "exists".to_owned(), + channel_id: channel_id.to_owned(), + resource_id: format!("res-{channel_id}"), + } +} + +#[tokio::test] +async fn publish_reaches_only_the_matching_token() { + let registry = RelayRegistry::default(); + let mut alpha = registry.subscribe("token-alpha"); + let mut beta = registry.subscribe("token-beta"); + + assert_eq!(registry.publish("token-alpha", ping("chan-1")), 1); + assert_eq!(alpha.recv().await.unwrap(), ping("chan-1")); + assert!(beta.try_recv().is_err()); +} + +#[tokio::test] +async fn publish_without_subscribers_is_dropped() { + let registry = RelayRegistry::default(); + assert_eq!(registry.publish("stray-token", ping("chan-1")), 0); +} + +#[tokio::test] +async fn disconnected_tokens_are_reclaimed_on_the_next_publish() { + let registry = RelayRegistry::default(); + let receiver = registry.subscribe("token-alpha"); + drop(receiver); + + assert_eq!(registry.publish("token-alpha", ping("chan-1")), 0); + assert!( + registry.inner.lock().unwrap().is_empty(), + "the dead entry is removed" + ); +} + +#[tokio::test] +async fn every_subscriber_for_one_token_receives_the_delivery() { + let registry = RelayRegistry::default(); + let mut first = registry.subscribe("token-alpha"); + let mut second = registry.subscribe("token-alpha"); + + assert_eq!(registry.publish("token-alpha", ping("chan-1")), 2); + assert_eq!(first.recv().await.unwrap(), ping("chan-1")); + assert_eq!(second.recv().await.unwrap(), ping("chan-1")); +} diff --git a/services/email_service/Cargo.toml b/services/email_service/Cargo.toml index 20788a13d7c..a4c9ed1674c 100644 --- a/services/email_service/Cargo.toml +++ b/services/email_service/Cargo.toml @@ -77,6 +77,7 @@ sfs_delete = [] [dependencies] calendar_events = { path = "../../crates/calendar_events", features = ["google", "inbound", "postgres"] } +calendar_watch_relay = { path = "../../crates/calendar_watch_relay" } ammonia = { workspace = true } anyhow = { workspace = true } authentication_service_client = { path = "../../crates/authentication_service_client" } diff --git a/services/email_service/src/bin/pubsub_workers/pubsub_workers.rs b/services/email_service/src/bin/pubsub_workers/pubsub_workers.rs index 37a7ff1d294..3a17daa756f 100644 --- a/services/email_service/src/bin/pubsub_workers/pubsub_workers.rs +++ b/services/email_service/src/bin/pubsub_workers/pubsub_workers.rs @@ -306,6 +306,14 @@ async fn main() -> anyhow::Result<()> { RateBudget::Backfill, ); + worker_tracker.spawn(email_service::pubsub::calendar_watch_subscriber::run( + db.clone(), + redis_client.clone(), + Arc::new(auth_service_client.clone()), + config.calendar_sync_enabled, + worker_cancellation_token.clone(), + )); + let sfs_client = StaticFileServiceClient::new( config.internal_api_key.to_string(), StaticFileServiceUrl::new()?.to_string(), diff --git a/services/email_service/src/openapi.rs b/services/email_service/src/openapi.rs index e0460e3524a..ee1a26bc5dc 100644 --- a/services/email_service/src/openapi.rs +++ b/services/email_service/src/openapi.rs @@ -4,6 +4,7 @@ mod api; mod backfill_completion_service; mod backfill_init_service; mod calendar_outbox; +mod calendar_tokens; mod config; mod outbound; mod pubsub; diff --git a/services/email_service/src/pubsub/calendar_watch_subscriber.rs b/services/email_service/src/pubsub/calendar_watch_subscriber.rs new file mode 100644 index 00000000000..1b6493dd8af --- /dev/null +++ b/services/email_service/src/pubsub/calendar_watch_subscriber.rs @@ -0,0 +1,238 @@ +//! Relay subscriber giving local stacks real Google Calendar push. +//! +//! When `CALENDAR_WATCH_RELAY_URL` is configured (local stacks only), this +//! worker connects OUT to the serving deployment's SSE endpoint and treats +//! each relayed notification exactly like a direct webhook delivery: verify +//! nothing (the relay already routed by this stack's own channel token) and +//! re-arm the watched inbox's sync job. The 5-minute poll remains the +//! backstop across disconnects, so every failure here degrades to today's +//! freshness instead of an error. +//! +//! On graceful shutdown the worker stops every open push channel at Google — +//! a local stack that is going away has no reason to keep deliveries flowing +//! toward the relay. Deployments never run this teardown: their channels are +//! meant to outlive any single replica. + +use std::sync::Arc; +use std::time::Duration; + +use authentication_service_client::AuthServiceClient; +use calendar_events::domain::service::{CalendarService, stop_all_watch_channels}; +use calendar_events::outbound::google::GoogleCalendarClient; +use calendar_events::outbound::pg::PgCalendarRepository; +use futures::StreamExt; +use sqlx::PgPool; +use tokio_util::sync::CancellationToken; + +use calendar_watch_relay::{ + RelayedWatchNotification, SseDataParser, WatchRelaySubscriberConfig, + watch_relay_subscriber_config, +}; + +use crate::calendar_tokens::CalendarTokenProviderAdapter; +use crate::pubsub::calendar_backfill_adapters::RedisCalendarRequestGate; +use crate::pubsub::context::calendar_watch_config; +use crate::util::redis::RedisClient; + +const INITIAL_BACKOFF: Duration = Duration::from_secs(1); +const MAX_BACKOFF: Duration = Duration::from_secs(60); +/// The server keep-alives every 15s, so a chunk gap this long means the +/// connection is dead even when TCP has not noticed yet. +const READ_TIMEOUT: Duration = Duration::from_secs(60); +/// Bound on the subscribe request itself: response headers must arrive +/// promptly even though the body then stays open indefinitely, so a stalled +/// server returns control to the reconnect loop instead of parking it. +const SUBSCRIBE_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); +/// Bound on the shutdown stop pass so container teardown is never hung. +const STOP_CHANNELS_TIMEOUT: Duration = Duration::from_secs(15); + +/// Subscribe to relayed watch notifications until cancelled, then stop the +/// stack's open push channels. A stack without relay configuration returns +/// immediately. +pub async fn run( + db: PgPool, + redis_client: RedisClient, + auth_service_client: Arc, + calendar_sync_enabled: bool, + cancellation: CancellationToken, +) { + let Some(config) = watch_relay_subscriber_config() else { + return; + }; + let Some(watch) = calendar_watch_config() else { + tracing::warn!( + "CALENDAR_WATCH_RELAY_URL is set without a complete watch config; not subscribing" + ); + return; + }; + if !calendar_sync_enabled { + tracing::warn!( + "CALENDAR_WATCH_RELAY_URL is set but calendar sync is disabled; not subscribing" + ); + return; + } + tracing::info!(url = %config.url, "subscribing to relayed calendar watch notifications"); + let calendar_service = CalendarService::new(PgCalendarRepository::new(db.clone())); + let client = match reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .build() + { + Ok(client) => client, + Err(error) => { + tracing::error!(error = ?error, "failed to build the relay subscriber http client"); + return; + } + }; + let mut backoff = INITIAL_BACKOFF; + loop { + tokio::select! { + () = cancellation.cancelled() => break, + connected = subscribe_once(&client, &config, &watch.token, &calendar_service) => { + match connected { + Ok(()) => { + backoff = INITIAL_BACKOFF; + tracing::info!("relay subscription ended; reconnecting"); + } + Err(error) => { + tracing::warn!(error = ?error, delay_secs = backoff.as_secs(), "relay subscription failed; backing off"); + } + } + } + } + tokio::select! { + () = cancellation.cancelled() => break, + () = tokio::time::sleep(backoff) => {} + } + backoff = (backoff * 2).min(MAX_BACKOFF); + } + stop_open_channels(db, redis_client, auth_service_client).await; +} + +/// Hold one SSE subscription, dispatching every relayed notification. +/// Returns `Ok` only after a healthy stream ends, so callers can distinguish +/// connection churn from setup failures when pacing reconnects. +async fn subscribe_once( + client: &reqwest::Client, + config: &WatchRelaySubscriberConfig, + token: &str, + calendar_service: &CalendarService, +) -> anyhow::Result<()> { + let request = client + .get(format!("{}/calendar/relay/subscribe", config.url)) + .header("x-relay-secret", &config.secret) + .header("x-relay-token", token) + .send(); + let response = tokio::time::timeout(SUBSCRIBE_REQUEST_TIMEOUT, request) + .await + .map_err(|_| anyhow::anyhow!("relay subscription request timed out"))??; + if response.status() != reqwest::StatusCode::OK { + anyhow::bail!( + "relay subscription rejected with status {}", + response.status() + ); + } + let mut stream = response.bytes_stream(); + let mut parser = SseDataParser::default(); + loop { + let chunk = match tokio::time::timeout(READ_TIMEOUT, stream.next()).await { + Ok(Some(Ok(chunk))) => chunk, + Ok(Some(Err(error))) => anyhow::bail!("relay stream failed: {error:?}"), + Ok(None) => return Ok(()), + Err(_) => anyhow::bail!("relay stream stalled past the keep-alive interval"), + }; + for payload in parser.push(&String::from_utf8_lossy(&chunk)) { + match serde_json::from_str::(&payload) { + Ok(notification) => apply(calendar_service, notification).await, + Err(error) => { + tracing::warn!(error = ?error, "undecodable relayed watch notification"); + } + } + } + } +} + +/// Mirror of the webhook handler's dispatch for one relayed notification. +async fn apply( + calendar_service: &CalendarService, + notification: RelayedWatchNotification, +) { + if notification.state == "sync" { + tracing::info!( + channel_id = %notification.channel_id, + "relayed watch channel handshake received" + ); + return; + } + match calendar_service + .handle_watch_notification(¬ification.channel_id, ¬ification.resource_id) + .await + { + Ok(matched) => { + if !matched { + tracing::debug!( + channel_id = %notification.channel_id, + "relayed calendar notification matched no active channel" + ); + } + } + Err(error) => { + tracing::warn!( + error = ?error, + channel_id = %notification.channel_id, + "failed to apply relayed calendar watch notification" + ); + } + } +} + +/// Best-effort stop of every open push channel, bounded so shutdown cannot +/// hang on a slow provider. +async fn stop_open_channels( + db: PgPool, + redis_client: RedisClient, + auth_service_client: Arc, +) { + let redis_conn = match redis_client.inner.get_multiplexed_async_connection().await { + Ok(connection) => connection, + Err(error) => { + tracing::warn!(error = ?error, "cannot stop watch channels without redis"); + return; + } + }; + let repository = PgCalendarRepository::new(db); + let provider_client = match reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + { + Ok(client) => client, + Err(error) => { + tracing::warn!(error = ?error, "failed to build the channel stop http client"); + return; + } + }; + let provider = GoogleCalendarClient::with_gate( + provider_client, + RedisCalendarRequestGate::new(redis_client), + ); + let tokens = CalendarTokenProviderAdapter::new(redis_conn, auth_service_client); + match tokio::time::timeout( + STOP_CHANNELS_TIMEOUT, + stop_all_watch_channels(&repository, &provider, &tokens), + ) + .await + { + Ok(Ok(summary)) => { + tracing::info!( + stopped = summary.stopped, + failed = summary.failed, + "stopped open calendar watch channels at shutdown" + ); + } + Ok(Err(error)) => { + tracing::warn!(error = ?error, "failed to stop calendar watch channels at shutdown"); + } + Err(_) => { + tracing::warn!("timed out stopping calendar watch channels at shutdown"); + } + } +} diff --git a/services/email_service/src/pubsub/mod.rs b/services/email_service/src/pubsub/mod.rs index e34bbceeef7..ad112cf6ab7 100644 --- a/services/email_service/src/pubsub/mod.rs +++ b/services/email_service/src/pubsub/mod.rs @@ -1,6 +1,8 @@ pub mod backfill; /// Outbound adapters used by calendar backfill application services. pub mod calendar_backfill_adapters; +/// Relay subscriber giving local stacks real Google Calendar push. +pub mod calendar_watch_subscriber; pub mod context; /// The flag-selected CRM metadata resolver, re-exported for the /// pubsub_workers binary to construct. diff --git a/tooling/just/xtask.just b/tooling/just/xtask.just index 5e3ecc5df3a..39626ca4fa8 100644 --- a/tooling/just/xtask.just +++ b/tooling/just/xtask.just @@ -36,6 +36,13 @@ destroy_local *ARGS: stop_local *ARGS: {{ xtask }} stop-local "$@" +# Toggle real Google Calendar push for a local instance (relayed via the dev +# email service): `just calendar-push enable|disable|status [--instance x]`. +# Requires CALENDAR_WATCH_RELAY_SECRET (Doppler or --env-file) to arm. +[positional-arguments] +calendar-push *ARGS: + @{{ xtask }} calendar-push "$@" + # Headless stack orchestration for previews, agents, and CI. # just stack up # full stack; static frontend served by the proxy # just stack update # rebuild + reload only what changed (the `r` hotkey) diff --git a/tooling/xtask/crates/xtask_local/src/local/calendar_push.rs b/tooling/xtask/crates/xtask_local/src/local/calendar_push.rs new file mode 100644 index 00000000000..228b8fc1fd4 --- /dev/null +++ b/tooling/xtask/crates/xtask_local/src/local/calendar_push.rs @@ -0,0 +1,117 @@ +//! Opt-in per-instance overlay pointing calendar watch channels at the dev +//! relay, so real Google Calendar push reaches a local stack. +//! +//! `enable` mints a per-instance channel token and writes `calendar-push.env` +//! into the instance's artifact dir; `env_layer::resolve` overlays that file +//! on every subsequent resolve (up, update, validate, snapshot), which keeps +//! every verb consistent without threading a flag through each of them. The +//! relay shared secret is intentionally NOT written here — it is a real +//! secret and must arrive through Doppler or `--env-file` as +//! `CALENDAR_WATCH_RELAY_SECRET`. + +use std::io::Read; +use std::path::PathBuf; + +use anyhow::{Context, Result, bail}; + +use super::instance::Instance; + +/// The dev-only tunnel service that receives Google's webhooks on the local +/// stack's behalf and relays them out (its hostname is a subdomain of +/// macro.com, already verified in the Cloud project owning the OAuth client, +/// which a laptop's never can be). +const DEV_TUNNEL: &str = "https://calendar-event-local-tunnel-dev.macro.com"; + +/// The overlay consumed by `env_layer::resolve` when present. +pub fn env_path(instance: &Instance) -> PathBuf { + instance.artifact_dir().join("calendar-push.env") +} + +fn token_path(instance: &Instance) -> PathBuf { + instance.artifact_dir().join("calendar-push.token") +} + +/// Mint the instance's channel token once and reuse it across enables, so +/// re-enabling never orphans channels opened under a previous token. Only a +/// genuinely absent file mints: any other read failure (or a corrupt empty +/// file) must not silently rotate the token out from under open channels. +fn ensure_token(instance: &Instance) -> Result { + let path = token_path(instance); + match std::fs::read_to_string(&path) { + Ok(existing) => { + let existing = existing.trim(); + if existing.is_empty() { + bail!( + "calendar push token file {} is empty — delete it to mint a fresh token", + path.display() + ); + } + return Ok(existing.to_owned()); + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(error).with_context(|| format!("reading {}", path.display())); + } + } + let mut bytes = [0u8; 32]; + std::fs::File::open("/dev/urandom") + .and_then(|mut file| file.read_exact(&mut bytes)) + .context("reading /dev/urandom for the calendar push token")?; + let token: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect(); + instance.ensure_artifact_dir()?; + std::fs::write(&path, &token).with_context(|| format!("writing {}", path.display()))?; + Ok(token) +} + +/// Write the overlay so the next resolve arms calendar push. +pub fn enable(instance: &Instance) -> Result<()> { + let token = ensure_token(instance)?; + let path = env_path(instance); + let contents = format!( + "# GENERATED by `just calendar-push enable` — remove via `just calendar-push disable`.\n\ + CALENDAR_WATCH_WEBHOOK_URL={DEV_TUNNEL}/calendar/notifications\n\ + CALENDAR_WATCH_RELAY_URL={DEV_TUNNEL}\n\ + CALENDAR_WATCH_TOKEN={token}\n" + ); + std::fs::write(&path, contents).with_context(|| format!("writing {}", path.display()))?; + println!( + "calendar push ENABLED for instance {} — takes effect on the next `just run_local` (or `just stack update`)", + instance.name() + ); + println!(" overlay {}", path.display()); + println!( + " requires CALENDAR_WATCH_RELAY_SECRET (Doppler or --env-file) and calendar sync enabled" + ); + Ok(()) +} + +/// Remove the overlay. The instance token is kept so a later enable reuses +/// it; a currently running stack keeps its channels until it shuts down +/// (graceful shutdown stops them at Google). +pub fn disable(instance: &Instance) -> Result<()> { + let path = env_path(instance); + if path.is_file() { + std::fs::remove_file(&path).with_context(|| format!("removing {}", path.display()))?; + println!( + "calendar push DISABLED for instance {} — takes effect on the next `just run_local` (or `just stack update`)", + instance.name() + ); + } else { + println!( + "calendar push already disabled for instance {}", + instance.name() + ); + } + Ok(()) +} + +/// Report whether the overlay is present. +pub fn status(instance: &Instance) -> Result<()> { + let path = env_path(instance); + if path.is_file() { + println!("calendar push: ENABLED ({})", path.display()); + } else { + println!("calendar push: disabled"); + } + Ok(()) +} diff --git a/tooling/xtask/crates/xtask_local/src/local/cli.rs b/tooling/xtask/crates/xtask_local/src/local/cli.rs index 4d207609e80..a3433ffb5cc 100644 --- a/tooling/xtask/crates/xtask_local/src/local/cli.rs +++ b/tooling/xtask/crates/xtask_local/src/local/cli.rs @@ -43,6 +43,9 @@ enum Cmd { SeedEnv(InstanceArgs), /// Run a seed scenario against an instance's host-facing endpoints. SeedScenario(SeedScenarioArgs), + /// Toggle real Google Calendar push for a local instance (relayed via + /// the dev email service). + CalendarPush(CalendarPushArgs), /// Stop an instance's containers (keep volumes). StopLocal(InstanceArgs), /// Drop, recreate, and migrate the instance database. @@ -180,6 +183,25 @@ pub struct ForceArg { pub force: bool, } +#[derive(Args, Clone)] +pub struct CalendarPushArgs { + #[command(flatten)] + pub instance: InstanceArgs, + /// What to do with the instance's calendar push overlay. + #[arg(value_enum)] + pub action: CalendarPushAction, +} + +#[derive(clap::ValueEnum, Clone, Copy, Debug)] +pub enum CalendarPushAction { + /// Write the overlay so the next stack (re)start arms push. + Enable, + /// Remove the overlay; a running stack keeps push until restarted. + Disable, + /// Report whether the overlay is present. + Status, +} + #[derive(Args, Clone)] #[command(trailing_var_arg = true)] pub struct SeedScenarioArgs { @@ -243,6 +265,17 @@ fn run(cli: Cli) -> Result<()> { let instance = super::instance::Instance::derive(a.instance.as_deref(), a.port_base)?; super::seed_env::emit(&instance) } + Cmd::CalendarPush(a) => { + let instance = super::instance::Instance::derive( + a.instance.instance.as_deref(), + a.instance.port_base, + )?; + match a.action { + CalendarPushAction::Enable => super::calendar_push::enable(&instance), + CalendarPushAction::Disable => super::calendar_push::disable(&instance), + CalendarPushAction::Status => super::calendar_push::status(&instance), + } + } Cmd::SeedScenario(a) => { let instance = super::instance::Instance::derive( a.instance.instance.as_deref(), diff --git a/tooling/xtask/crates/xtask_local/src/local/env_layer.rs b/tooling/xtask/crates/xtask_local/src/local/env_layer.rs index ef6bd866750..9f827204c5f 100644 --- a/tooling/xtask/crates/xtask_local/src/local/env_layer.rs +++ b/tooling/xtask/crates/xtask_local/src/local/env_layer.rs @@ -64,6 +64,7 @@ pub fn resolve( if let Some(local) = &local { env.extend(local.boot_stub_env()); } + let mut calendar_push_overlay: Option> = None; let doppler_used = if no_doppler { false } else { @@ -73,6 +74,18 @@ pub fn resolve( for (k, v) in local.to_env() { env.insert(k, v); } + // Opt-in calendar push: `just calendar-push enable` materializes this + // overlay, and every resolve (up, update, validate, snapshot) picks it + // up here so the verbs can never disagree about whether push is on. + let calendar_push = super::calendar_push::env_path(instance); + if calendar_push.is_file() { + let mut overlay = BTreeMap::new(); + load_dotenv_into(&calendar_push, &mut overlay).with_context(|| { + format!("loading calendar push overlay {}", calendar_push.display()) + })?; + env.extend(overlay.clone()); + calendar_push_overlay = Some(overlay); + } // Opt-in local trace export: point services at the local OTLP collector // (docker-network alias `otel-collector`) only when one answers on the // OTLP HTTP port, so services don't spam export errors when none is @@ -115,6 +128,33 @@ pub fn resolve( pull_aws_credentials(&mut env); } + // The relay secret rides a later layer (Doppler or --env-file), so only + // the fully merged map can tell whether calendar push will actually arm. + // Without a usable secret the overlay must leave no trace: a lone + // CALENDAR_WATCH_WEBHOOK_URL/TOKEN pair would open watch channels nobody + // subscribes to. + if !calendar_push_secret_ok(&env) { + if let Some(overlay) = &calendar_push_overlay { + strip_calendar_push_overlay(&mut env, overlay); + eprintln!( + "warning: calendar push is enabled but CALENDAR_WATCH_RELAY_SECRET is missing \ + or blank — disarming calendar push for this run; supply the secret via \ + Doppler or --env-file" + ); + } else if env + .get("CALENDAR_WATCH_RELAY_URL") + .is_some_and(|url| !url.trim().is_empty()) + { + // Hand-armed relay config (Doppler or --env-file) is the + // developer's to keep, so only warn. + eprintln!( + "warning: CALENDAR_WATCH_RELAY_URL is set but CALENDAR_WATCH_RELAY_SECRET is \ + missing or blank — the stack will open watch channels without subscribing \ + to their deliveries" + ); + } + } + let generated_path = instance.ensure_artifact_dir()?.join("local.generated.env"); write_dotenv(&generated_path, &env) .with_context(|| format!("writing {}", generated_path.display()))?; @@ -131,6 +171,27 @@ fn should_overlay_process_env(key: &str, value: &str) -> bool { !(key == "DATABASE_URL" && is_local_database_url(value)) } +/// Whether the merged env carries a usable relay secret. Blank counts as +/// absent, matching how the service reads every watch variable. +fn calendar_push_secret_ok(env: &BTreeMap) -> bool { + env.get("CALENDAR_WATCH_RELAY_SECRET") + .is_some_and(|secret| !secret.trim().is_empty()) +} + +/// Remove what the calendar push overlay contributed. A key whose merged +/// value no longer matches the overlay was overridden by a later layer on +/// purpose, so it stays. +fn strip_calendar_push_overlay( + env: &mut BTreeMap, + overlay: &BTreeMap, +) { + for (key, value) in overlay { + if env.get(key) == Some(value) { + env.remove(key); + } + } +} + pub(super) fn is_local_database_url(value: &str) -> bool { value.contains("@postgres:") || value.contains("localhost") } diff --git a/tooling/xtask/crates/xtask_local/src/local/env_layer/test.rs b/tooling/xtask/crates/xtask_local/src/local/env_layer/test.rs index 4b600ae898b..0908f2c4aeb 100644 --- a/tooling/xtask/crates/xtask_local/src/local/env_layer/test.rs +++ b/tooling/xtask/crates/xtask_local/src/local/env_layer/test.rs @@ -19,3 +19,39 @@ fn still_allows_non_local_database_overrides() { "postgres://user:password@dev.example.com:5432/macrodb" )); } + +#[test] +fn blank_relay_secret_counts_as_absent() { + let mut env = BTreeMap::new(); + assert!(!calendar_push_secret_ok(&env)); + env.insert("CALENDAR_WATCH_RELAY_SECRET".into(), " ".into()); + assert!(!calendar_push_secret_ok(&env)); + env.insert("CALENDAR_WATCH_RELAY_SECRET".into(), "s3cret".into()); + assert!(calendar_push_secret_ok(&env)); +} + +#[test] +fn stripping_the_overlay_keeps_later_layer_overrides() { + let overlay: BTreeMap = [ + ("CALENDAR_WATCH_WEBHOOK_URL", "https://dev/notifications"), + ("CALENDAR_WATCH_RELAY_URL", "https://dev"), + ("CALENDAR_WATCH_TOKEN", "overlay-token"), + ] + .into_iter() + .map(|(k, v)| (k.to_owned(), v.to_owned())) + .collect(); + let mut env = overlay.clone(); + env.insert("CALENDAR_WATCH_TOKEN".into(), "env-file-token".into()); + env.insert("UNRELATED".into(), "kept".into()); + + strip_calendar_push_overlay(&mut env, &overlay); + + assert!(!env.contains_key("CALENDAR_WATCH_WEBHOOK_URL")); + assert!(!env.contains_key("CALENDAR_WATCH_RELAY_URL")); + assert_eq!( + env.get("CALENDAR_WATCH_TOKEN").map(String::as_str), + Some("env-file-token"), + "an --env-file override is the developer's choice and survives" + ); + assert_eq!(env.get("UNRELATED").map(String::as_str), Some("kept")); +} diff --git a/tooling/xtask/crates/xtask_local/src/local/mod.rs b/tooling/xtask/crates/xtask_local/src/local/mod.rs index cdbcfd1dcf0..dcdc389bdd4 100644 --- a/tooling/xtask/crates/xtask_local/src/local/mod.rs +++ b/tooling/xtask/crates/xtask_local/src/local/mod.rs @@ -15,6 +15,7 @@ use std::path::PathBuf; pub mod arch; pub mod build; +pub mod calendar_push; pub mod cli; pub mod db; pub mod docker;