Merge branch 'master' into sessions-global-in

frontend/src/lib/integrations/IntegrationScopesWarning.tsx

@@ -0,0 +1,65 @@
+import api from 'lib/api'
+import { LemonBanner } from 'lib/lemon-ui/LemonBanner'
+import { Link } from 'lib/lemon-ui/Link'
+import { useMemo } from 'react'
+
+import { HogFunctionInputSchemaType, IntegrationType } from '~/types'
+
+export function IntegrationScopesWarning({
+    integration,
+    schema,
+}: {
+    integration: IntegrationType
+    schema?: HogFunctionInputSchemaType
+}): JSX.Element {
+    const getScopes = useMemo((): string[] => {
+        const scopes: any[] = []
+        const possibleScopeLocation = [integration.config.scope, integration.config.scopes]
+
+        possibleScopeLocation.map((scope) => {
+            if (typeof scope === 'string') {
+                scopes.push(scope.split(' '))
+                scopes.push(scope.split(','))
+            }
+            if (typeof scope === 'object') {
+                scopes.push(scope)
+            }
+        })
+        return scopes
+            .filter((scope: any) => typeof scope === 'object')
+            .reduce((a, b) => (a.length > b.length ? a : b), [])
+    }, [integration.config])
+
+    const requiredScopes = schema?.requiredScopes?.split(' ') || []
+    const missingScopes = requiredScopes.filter((scope: string) => !getScopes.includes(scope))
+
+    if (missingScopes.length === 0 || getScopes.length === 0) {
+        return <></>
+    }
+    return (
+        <div className="p-2">
+            <LemonBanner
+                type="error"
+                action={{
+                    children: 'Reconnect',
+                    disableClientSideRouting: true,
+                    to: api.integrations.authorizeUrl({
+                        kind: integration.kind,
+                        next: window.location.pathname,
+                    }),
+                }}
+            >
+                <span>Required scopes are missing: [{missingScopes.join(', ')}].</span>
+                {integration.kind === 'hubspot' ? (
+                    <span>
+                        Note that some features may not be available on your current HubSpot plan. Check out{' '}
+                        <Link to="https://developers.hubspot.com/beta-docs/guides/apps/authentication/scopes">
+                            this page
+                        </Link>{' '}
+                        for more details.
+                    </span>
+                ) : null}
+            </LemonBanner>
+        </div>
+    )
+}

frontend/src/lib/integrations/IntegrationView.tsx

@@ -1,15 +1,18 @@
 import { LemonBanner } from '@posthog/lemon-ui'
 import api from 'lib/api'
 import { UserActivityIndicator } from 'lib/components/UserActivityIndicator/UserActivityIndicator'
+import { IntegrationScopesWarning } from 'lib/integrations/IntegrationScopesWarning'
 
-import { IntegrationType } from '~/types'
+import { HogFunctionInputSchemaType, IntegrationType } from '~/types'
 
 export function IntegrationView({
     integration,
     suffix,
+    schema,
 }: {
     integration: IntegrationType
     suffix?: JSX.Element
+    schema?: HogFunctionInputSchemaType
 }): JSX.Element {
     const errors = (integration.errors && integration.errors?.split(',')) || []
 
@@ -36,7 +39,7 @@ export function IntegrationView({
                 {suffix}
             </div>
 
-            {errors.length > 0 && (
+            {errors.length > 0 ? (
                 <div className="p-2">
                     <LemonBanner
                         type="error"
@@ -54,6 +57,8 @@ export function IntegrationView({
                             : `There was an error with this integration: ${errors[0]}`}
                     </LemonBanner>
                 </div>
+            ) : (
+                <IntegrationScopesWarning integration={integration} schema={schema} />
             )}
         </div>
     )

frontend/src/scenes/pipeline/hogfunctions/integrations/HogFunctionInputIntegration.tsx

@@ -16,6 +16,7 @@ export function HogFunctionInputIntegration({ schema, ...props }: HogFunctionInp
         <>
             <IntegrationChoice
                 {...props}
+                schema={schema}
                 integration={schema.integration}
                 redirectUrl={`${window.location.pathname}?integration_target=${schema.key}`}
                 beforeRedirect={() => persistForUnload()}

frontend/src/scenes/pipeline/hogfunctions/integrations/IntegrationChoice.tsx

@@ -7,17 +7,21 @@ import { IntegrationView } from 'lib/integrations/IntegrationView'
 import { capitalizeFirstLetter } from 'lib/utils'
 import { urls } from 'scenes/urls'
 
+import { HogFunctionInputSchemaType } from '~/types'
+
 export type IntegrationConfigureProps = {
     value?: number
     onChange?: (value: number | null) => void
     redirectUrl?: string
+    schema?: HogFunctionInputSchemaType
     integration?: string
     beforeRedirect?: () => void
 }
 
 export function IntegrationChoice({
     onChange,
     value,
+    schema,
     integration,
     redirectUrl,
     beforeRedirect,
@@ -124,5 +128,13 @@ export function IntegrationChoice({
         </LemonMenu>
     )
 
-    return <>{integrationKind ? <IntegrationView integration={integrationKind} suffix={button} /> : button}</>
+    return (
+        <>
+            {integrationKind ? (
+                <IntegrationView schema={schema} integration={integrationKind} suffix={button} />
+            ) : (
+                button
+            )}
+        </>
+    )
 }

frontend/src/types.ts

@@ -4605,6 +4605,7 @@ export type HogFunctionInputSchemaType = {
     integration?: string
     integration_key?: string
     integration_field?: 'slack_channel'
+    requiredScopes?: string
 }
 
 export type HogFunctionInputType = {

plugin-server/src/cdp/types.ts

@@ -272,6 +272,7 @@ export type HogFunctionInputSchemaType = {
     integration?: string
     integration_key?: string
     integration_field?: 'slack_channel'
+    requiredScopes?: string
 }
 
 export type HogFunctionTypeType = 'destination' | 'email' | 'sms' | 'push' | 'activity' | 'alert' | 'broadcast'

posthog/api/survey.py

@@ -325,7 +325,7 @@ def validate(self, data):
             if response_sampling_start_date < today_utc:
                 raise serializers.ValidationError(
                     {
-                        "response_sampling_start_date": "Response sampling start date must be today or a future date in UTC."
+                        "response_sampling_start_date": f"Response sampling start date must be today or a future date in UTC. Got {response_sampling_start_date} when current time is {today_utc}"
                     }
                 )
 

posthog/api/test/test_survey.py

@@ -2378,6 +2378,7 @@ def test_can_clear_associated_actions(self):
         assert len(survey.actions.all()) == 0
 
 
+@freeze_time("2024-12-12 00:00:00")
 class TestSurveyResponseSampling(APIBaseTest):
     def _create_survey_with_sampling_limits(
         self,
@@ -2407,6 +2408,7 @@ def _create_survey_with_sampling_limits(
         )
 
         response_data = response.json()
+        assert response.status_code == status.HTTP_201_CREATED, response_data
         survey = Survey.objects.get(id=response_data["id"])
         return survey
 

posthog/batch_exports/service.py

@@ -794,3 +794,19 @@ async def aupdate_batch_export_backfill_status(backfill_id: UUID, status: str) -
         raise ValueError(f"BatchExportBackfill with id {backfill_id} not found.")
 
     return await model.aget()
+
+
+async def aupdate_records_total_count(
+    batch_export_id: UUID, interval_start: dt.datetime, interval_end: dt.datetime, count: int
+) -> int:
+    """Update the expected records count for a set of batch export runs.
+
+    Typically, there is one batch export run per batch export interval, however
+    there could be multiple if data has been backfilled.
+    """
+    rows_updated = await BatchExportRun.objects.filter(
+        batch_export_id=batch_export_id,
+        data_interval_start=interval_start,
+        data_interval_end=interval_end,
+    ).aupdate(records_total_count=count)
+    return rows_updated

posthog/batch_exports/sql.py

@@ -318,3 +318,22 @@
     SETTINGS optimize_aggregation_in_order=1
 )
 """
+
+# TODO: is this the best query to use?
+EVENT_COUNT_BY_INTERVAL = """
+SELECT
+    toStartOfInterval(_inserted_at, INTERVAL {interval}) AS interval_start,
+    interval_start + INTERVAL {interval} AS interval_end,
+    COUNT(*) as total_count
+FROM
+    events_batch_export_recent(
+        team_id={team_id},
+        interval_start={overall_interval_start},
+        interval_end={overall_interval_end},
+        include_events={include_events}::Array(String),
+        exclude_events={exclude_events}::Array(String)
+    ) AS events
+GROUP BY interval_start
+ORDER BY interval_start desc
+SETTINGS max_replica_delay_for_distributed_queries=1
+"""

posthog/cdp/templates/google_ads/template_google_ads.py

@@ -55,6 +55,7 @@
             "type": "integration",
             "integration": "google-ads",
             "label": "Google Ads account",
+            "requiredScopes": "https://www.googleapis.com/auth/adwords https://www.googleapis.com/auth/userinfo.email",
             "secret": False,
             "required": True,
         },

posthog/cdp/templates/hubspot/template_hubspot.py

@@ -61,6 +61,7 @@
             "type": "integration",
             "integration": "hubspot",
             "label": "Hubspot connection",
+            "requiredScopes": "crm.objects.contacts.write crm.objects.contacts.read",
             "secret": False,
             "required": True,
         },
@@ -307,6 +308,7 @@
             "type": "integration",
             "integration": "hubspot",
             "label": "Hubspot connection",
+            "requiredScopes": "analytics.behavioral_events.send behavioral_events.event_definitions.read_write",
             "secret": False,
             "required": True,
         },

posthog/cdp/templates/salesforce/template_salesforce.py

@@ -15,6 +15,7 @@
         "type": "integration",
         "integration": "salesforce",
         "label": "Salesforce account",
+        "requiredScopes": "refresh_token full",
         "secret": False,
         "required": True,
     }

posthog/cdp/templates/slack/template_slack.py

@@ -34,6 +34,7 @@
             "type": "integration",
             "integration": "slack",
             "label": "Slack workspace",
+            "requiredScopes": "channels:read groups:read chat:write chat:write.customize",
             "secret": False,
             "required": True,
         },

posthog/cdp/validation.py

@@ -65,6 +65,7 @@ class InputsSchemaItemSerializer(serializers.Serializer):
     integration = serializers.CharField(required=False)
     integration_key = serializers.CharField(required=False)
     integration_field = serializers.ChoiceField(choices=["slack_channel"], required=False)
+    requiredScopes = serializers.CharField(required=False)
 
     # TODO Validate choices if type=choice
 

posthog/models/integration.py

@@ -163,7 +163,7 @@ def oauth_config_for_kind(cls, kind: str) -> OauthConfig:
                 authorize_url="https://app.hubspot.com/oauth/authorize",
                 token_url="https://api.hubapi.com/oauth/v1/token",
                 token_info_url="https://api.hubapi.com/oauth/v1/access-tokens/:access_token",
-                token_info_config_fields=["hub_id", "hub_domain", "user", "user_id"],
+                token_info_config_fields=["hub_id", "hub_domain", "user", "user_id", "scopes"],
                 client_id=settings.HUBSPOT_APP_CLIENT_ID,
                 client_secret=settings.HUBSPOT_APP_CLIENT_SECRET,
                 scope="tickets crm.objects.contacts.write sales-email-read crm.objects.companies.read crm.objects.deals.read crm.objects.contacts.read crm.objects.quotes.read crm.objects.companies.write",
@@ -187,7 +187,7 @@ def oauth_config_for_kind(cls, kind: str) -> OauthConfig:
                 token_url="https://oauth2.googleapis.com/token",
                 client_id=settings.SOCIAL_AUTH_GOOGLE_OAUTH2_KEY,
                 client_secret=settings.SOCIAL_AUTH_GOOGLE_OAUTH2_SECRET,
-                scope="https://www.googleapis.com/auth/adwords email",
+                scope="https://www.googleapis.com/auth/adwords https://www.googleapis.com/auth/userinfo.email",
                 id_path="sub",
                 name_path="email",
             )

posthog/models/test/test_integration_model.py

@@ -120,7 +120,7 @@ def test_authorize_url_with_additional_authorize_params(self):
             url = OauthIntegration.authorize_url("google-ads", next="/projects/test")
             assert (
                 url
-                == "https://accounts.google.com/o/oauth2/v2/auth?client_id=google-client-id&scope=https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fadwords+email&redirect_uri=https%3A%2F%2Flocalhost%3A8000%2Fintegrations%2Fgoogle-ads%2Fcallback&response_type=code&state=next%3D%252Fprojects%252Ftest&access_type=offline&prompt=consent"
+                == "https://accounts.google.com/o/oauth2/v2/auth?client_id=google-client-id&scope=https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fadwords+https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fuserinfo.email&redirect_uri=https%3A%2F%2Flocalhost%3A8000%2Fintegrations%2Fgoogle-ads%2Fcallback&response_type=code&state=next%3D%252Fprojects%252Ftest&access_type=offline&prompt=consent"
             )
 
     @patch("posthog.models.integration.requests.post")
@@ -199,6 +199,10 @@ def test_integration_fetches_info_from_token_info_url(self, mock_get, mock_post)
                 "user": "user",
                 "user_id": "user_id",
                 "should_not": "be_saved",
+                "scopes": [
+                    "crm.objects.contacts.read",
+                    "crm.objects.contacts.write",
+                ],
             }
 
             with freeze_time("2024-01-01T12:00:00Z"):
@@ -219,6 +223,10 @@ def test_integration_fetches_info_from_token_info_url(self, mock_get, mock_post)
                 "user": "user",
                 "user_id": "user_id",
                 "refreshed_at": 1704110400,
+                "scopes": [
+                    "crm.objects.contacts.read",
+                    "crm.objects.contacts.write",
+                ],
             }
             assert integration.sensitive_config == {
                 "access_token": "FAKES_ACCESS_TOKEN",

posthog/temporal/batch_exports/__init__.py

@@ -17,6 +17,12 @@
     HttpBatchExportWorkflow,
     insert_into_http_activity,
 )
+from posthog.temporal.batch_exports.monitoring import (
+    BatchExportMonitoringWorkflow,
+    get_batch_export,
+    get_event_counts,
+    update_batch_export_runs,
+)
 from posthog.temporal.batch_exports.noop import NoOpWorkflow, noop_activity
 from posthog.temporal.batch_exports.postgres_batch_export import (
     PostgresBatchExportWorkflow,
@@ -54,6 +60,7 @@
     SnowflakeBatchExportWorkflow,
     HttpBatchExportWorkflow,
     SquashPersonOverridesWorkflow,
+    BatchExportMonitoringWorkflow,
 ]
 
 ACTIVITIES = [
@@ -76,4 +83,7 @@
     update_batch_export_backfill_model_status,
     wait_for_mutation,
     wait_for_table,
+    get_batch_export,
+    get_event_counts,
+    update_batch_export_runs,
 ]

posthog/temporal/batch_exports/bigquery_batch_export.py

@@ -391,7 +391,17 @@ async def amerge_person_tables(
 
         merge_query = f"""
         MERGE `{final_table.full_table_id.replace(":", ".", 1)}` final
-        USING `{stage_table.full_table_id.replace(":", ".", 1)}` stage
+        USING (
+            SELECT * FROM
+            (
+              SELECT
+              *,
+              ROW_NUMBER() OVER (PARTITION BY {",".join(field.name for field in merge_key)}) row_num
+            FROM
+              `{stage_table.full_table_id.replace(":", ".", 1)}`
+            )
+            WHERE row_num = 1
+        ) stage
         {merge_condition}
 
         WHEN MATCHED AND (stage.`{person_version_key}` > final.`{person_version_key}` OR stage.`{person_distinct_id_version_key}` > final.`{person_distinct_id_version_key}`) THEN