integrations gcs and s3

.env.example

@@ -4,10 +4,21 @@ DATABASE_URL="postgresql://postgres:postgres@localhost:5432/postgres?schema=publ
 DATABASE_URL_DIRECT="$DATABASE_URL"
 CLICKHOUSE_URL="http://localhost:8123/openpanel"
 
-# Symmetric key used to encrypt TOTP secrets and other sensitive data at rest.
+# Symmetric key for all at-rest encryption: TOTP secrets, GSC tokens, and
+# object-store export credentials (S3 secret keys, GCS service-account keys).
 # Generate with: openssl rand -hex 32
 ENCRYPTION_KEY=""
 
+# OBJECT-STORE EXPORT (S3/GCS) tuning — optional, sensible defaults shown.
+# The flushExports cron job windows ClickHouse by inserted_at and uploads
+# batched files. LAG keeps a safety gap behind now() for in-flight inserts;
+# BATCH_SIZE is rows per file; MAX_BATCHES_PER_RUN bounds backlog drain per
+# tick; CONCURRENCY is parallel (project,integration) uploads.
+# EXPORT_LAG_SECONDS="60"
+# EXPORT_BATCH_SIZE="50000"
+# EXPORT_MAX_BATCHES_PER_RUN="20"
+# EXPORT_CONCURRENCY="4"
+
 # REST
 BATCH_SIZE="5000"
 BATCH_INTERVAL="10000"

apps/api/package.json

@@ -77,5 +77,10 @@
     "tsdown": "0.14.2",
     "typescript": "catalog:",
     "vitest": "^1.0.0"
+  },
+  "peerDependencies": {
+    "@aws-sdk/client-s3": "^3.974.0",
+    "@aws-sdk/client-sts": "^3.974.0",
+    "@google-cloud/storage": "^7.18.0"
   }
 }

apps/api/src/controllers/webhook.controller.ts

@@ -25,6 +25,9 @@ const paramsSchema = z.object({
 const metadataSchema = z.object({
   organizationId: z.string(),
   integrationId: z.string(),
+  // Optional for back-compat with install URLs generated before integrations
+  // became project-scoped; the post-install redirect falls back to the org page.
+  projectId: z.string().optional(),
 });
 
 export async function slackWebhook(
@@ -85,7 +88,7 @@ export async function slackWebhook(
         '👋 Hello. You have successfully connected OpenPanel.dev to your Slack workspace.',
     });
 
-    const { organizationId, integrationId } = parsedMetadata.data;
+    const { organizationId, integrationId, projectId } = parsedMetadata.data;
 
     await db.integration.update({
       where: {
@@ -100,8 +103,16 @@ export async function slackWebhook(
       },
     });
 
+    const dashboardUrl =
+      process.env.DASHBOARD_URL || process.env.NEXT_PUBLIC_DASHBOARD_URL;
+    // Integrations are project-scoped; the org-level integrations route no longer
+    // exists. Newer installs carry projectId in their metadata. Older in-flight
+    // installs (started before the project-scoped routes shipped) may lack it —
+    // fall back to the org landing page rather than a now-404 integrations URL.
     return reply.redirect(
-      `${process.env.DASHBOARD_URL || process.env.NEXT_PUBLIC_DASHBOARD_URL}/${organizationId}/integrations/installed`
+      projectId
+        ? `${dashboardUrl}/${organizationId}/${projectId}/integrations/installed`
+        : `${dashboardUrl}/${organizationId}`
     );
   } catch (err) {
     request.log.error(err);

apps/api/tsdown.config.ts

@@ -10,6 +10,10 @@ const options: Options = {
     'pino',
     'pino-pretty',
     '@node-rs/argon2',
+    // integrations package
+    '@aws-sdk/client-s3',
+    '@aws-sdk/client-sts',
+    '@google-cloud/storage',
   ],
   sourcemap: true,
   platform: 'node',

apps/start/src/components/integrations/active-integrations.tsx

@@ -17,11 +17,11 @@ import {
 import { INTEGRATIONS } from './integrations';
 
 export function ActiveIntegrations() {
-  const { organizationId } = useAppParams();
+  const { projectId } = useAppParams();
   const trpc = useTRPC();
   const query = useQuery(
     trpc.integration.list.queryOptions({
-      organizationId: organizationId!,
+      projectId: projectId!,
     }),
   );
   const client = useQueryClient();
@@ -30,7 +30,7 @@ export function ActiveIntegrations() {
       onSuccess() {
         client.refetchQueries(
           trpc.integration.list.queryFilter({
-            organizationId,
+            projectId,
           }),
         );
       },

apps/start/src/components/integrations/forms/discord-integration.tsx

@@ -4,7 +4,6 @@ import { useAppParams } from '@/hooks/use-app-params';
 import { useTRPC } from '@/integrations/trpc/react';
 import type { RouterOutputs } from '@/trpc/client';
 import { zodResolver } from '@hookform/resolvers/zod';
-import { sendTestDiscordNotification } from '@openpanel/integrations/src/discord';
 import { zCreateDiscordIntegration } from '@openpanel/validation';
 import { useMutation } from '@tanstack/react-query';
 import { path, mergeDeepRight } from 'ramda';
@@ -21,12 +20,12 @@ export function DiscordIntegrationForm({
   defaultValues?: RouterOutputs['integration']['get'];
   onSuccess: () => void;
 }) {
-  const { organizationId } = useAppParams();
+  const { projectId } = useAppParams();
   const form = useForm<IForm>({
     defaultValues: mergeDeepRight(
       {
         id: defaultValues?.id,
-        organizationId,
+        projectId,
         config: {
           type: 'discord' as const,
           url: '',
@@ -55,13 +54,19 @@ export function DiscordIntegrationForm({
     toast.error('Validation error');
   };
 
+  const testMutation = useMutation(
+    trpc.integration.testConnection.mutationOptions(),
+  );
+
   const handleTest = async () => {
-    const webhookUrl = form.getValues('config.url');
-    if (!webhookUrl) {
+    const url = form.getValues('config.url');
+    if (!url) {
       return toast.error('Webhook URL is required');
     }
-    const res = await sendTestDiscordNotification(webhookUrl);
-    if (res.ok) {
+    const res = await testMutation.mutateAsync({
+      config: { type: 'discord', url },
+    });
+    if (res.success) {
       toast.success('Test notification sent');
     } else {
       toast.error('Failed to send test notification');

apps/start/src/components/integrations/forms/gcs-export-integration.tsx

@@ -0,0 +1,185 @@
+import { InputWithLabel } from '@/components/forms/input-with-label';
+import { Button } from '@/components/ui/button';
+import {
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectTrigger,
+  SelectValue,
+} from '@/components/ui/select';
+import { useAppParams } from '@/hooks/use-app-params';
+import { useTRPC } from '@/integrations/trpc/react';
+import type { RouterOutputs } from '@/trpc/client';
+import { zodResolver } from '@hookform/resolvers/zod';
+import { zCreateGCSExportIntegration } from '@openpanel/validation';
+import { useMutation } from '@tanstack/react-query';
+import { path, mergeDeepRight } from 'ramda';
+import { Controller, useForm } from 'react-hook-form';
+import { toast } from 'sonner';
+import type { z } from 'zod';
+
+type IForm = z.infer<typeof zCreateGCSExportIntegration>;
+
+export function GCSExportIntegrationForm({
+  defaultValues,
+  onSuccess,
+}: {
+  defaultValues?: RouterOutputs['integration']['get'];
+  onSuccess: () => void;
+}) {
+  const { projectId } = useAppParams();
+  const form = useForm<IForm>({
+    defaultValues: mergeDeepRight(
+      {
+        id: defaultValues?.id,
+        projectId,
+        name: '',
+        config: {
+          type: 'gcs_export' as const,
+          bucket: '',
+          prefix: 'openpanel-exports',
+          format: 'jsonl_gzip' as const,
+          serviceAccountKey: '',
+        },
+      },
+      defaultValues ?? {},
+    ),
+    resolver: zodResolver(zCreateGCSExportIntegration),
+  });
+  const trpc = useTRPC();
+  const mutation = useMutation(
+    trpc.integration.createOrUpdateExport.mutationOptions({
+      onSuccess,
+      onError(error) {
+        toast.error(error.message || 'Failed to create integration');
+      },
+    }),
+  );
+
+  const testMutation = useMutation(
+    trpc.integration.testExportConnection.mutationOptions({
+      onSuccess(data) {
+        if (data.success) {
+          toast.success('Connection successful! Bucket is accessible.');
+        } else {
+          toast.error(`Connection failed: ${data.error}`);
+        }
+      },
+      onError(error) {
+        toast.error(error.message || 'Failed to test connection');
+      },
+    }),
+  );
+
+  const handleSubmit = (values: IForm) => {
+    mutation.mutate(values);
+  };
+
+  const handleError = () => {
+    toast.error('Please fix validation errors');
+  };
+
+  const handleTest = () => {
+    const values = form.getValues();
+    if (!values.config.bucket || !values.config.serviceAccountKey) {
+      return toast.error('Bucket and Service Account Key are required');
+    }
+    testMutation.mutate(values);
+  };
+
+  return (
+    <form
+      onSubmit={form.handleSubmit(handleSubmit, handleError)}
+      className="col gap-4"
+    >
+      <InputWithLabel
+        label="Name"
+        placeholder="Eg. Production GCS Export"
+        {...form.register('name')}
+        error={form.formState.errors.name?.message}
+      />
+
+      <div className="grid grid-cols-2 gap-4">
+        <InputWithLabel
+          label="GCS Bucket"
+          placeholder="my-analytics-bucket"
+          {...form.register('config.bucket')}
+          error={path(['config', 'bucket', 'message'], form.formState.errors)}
+        />
+        <InputWithLabel
+          label="Prefix"
+          placeholder="openpanel-exports"
+          {...form.register('config.prefix')}
+          error={path(['config', 'prefix', 'message'], form.formState.errors)}
+        />
+      </div>
+
+      <div className="col gap-1.5">
+        <label className="text-sm font-medium">Format</label>
+        <Controller
+          name="config.format"
+          control={form.control}
+          render={({ field }) => (
+            <Select onValueChange={field.onChange} value={field.value}>
+              <SelectTrigger className="w-48">
+                <SelectValue placeholder="Select format" />
+              </SelectTrigger>
+              <SelectContent>
+                <SelectItem value="jsonl_gzip">JSONL (gzip)</SelectItem>
+                <SelectItem value="parquet" disabled>
+                  Parquet (coming soon)
+                </SelectItem>
+              </SelectContent>
+            </Select>
+          )}
+        />
+      </div>
+
+      <div className="col gap-1.5">
+        <label className="text-sm font-medium">
+          Service Account Key (JSON)
+        </label>
+        <textarea
+          className="border-input bg-background ring-offset-background placeholder:text-muted-foreground focus-visible:ring-ring min-h-32 w-full rounded-md border px-3 py-2 text-sm focus-visible:ring-2 focus-visible:ring-offset-2 focus-visible:outline-none disabled:cursor-not-allowed disabled:opacity-50"
+          placeholder='{"type": "service_account", "project_id": "...", ...}'
+          {...form.register('config.serviceAccountKey')}
+        />
+        {!!path(
+          ['config', 'serviceAccountKey', 'message'],
+          form.formState.errors,
+        ) && (
+          <p className="text-destructive text-xs">
+            {
+              path(
+                ['config', 'serviceAccountKey', 'message'],
+                form.formState.errors,
+              ) as any
+            }
+          </p>
+        )}
+        <p className="text-muted-foreground text-xs">
+          Paste the contents of your GCS service account JSON key file. The
+          service account needs write access to the specified bucket.
+        </p>
+      </div>
+
+      <div className="row gap-4">
+        <Button
+          type="button"
+          variant="outline"
+          onClick={handleTest}
+          disabled={testMutation.isPending}
+        >
+          {testMutation.isPending ? 'Testing...' : 'Test connection'}
+        </Button>
+        <Button type="submit" className="flex-1" disabled={mutation.isPending}>
+          {mutation.isPending
+            ? 'Saving...'
+            : defaultValues?.id
+              ? 'Update'
+              : 'Create'}
+        </Button>
+      </div>
+    </form>
+  );
+}