PROTM-2269: Add GCS HMAC connector distinction + GCS sink destination

src/cli/commands/migrate.test.ts

@@ -1942,6 +1942,103 @@ EXPORT_COMPRESSION gzip
     expect(output).toContain('compression: "gzip"');
   });
 
+  it("migrates GCS HMAC sink pipes and connections to TypeScript", async () => {
+    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "tinybird-migrate-"));
+    tempDirs.push(tempDir);
+
+    writeFile(
+      tempDir,
+      "exports_gcs_hmac.connection",
+      `TYPE gcs_hmac
+GCS_HMAC_ACCESS_ID {{ tb_secret("GCS_ACCESS_ID") }}
+GCS_HMAC_SECRET {{ tb_secret("GCS_SECRET") }}
+`
+    );
+
+    writeFile(
+      tempDir,
+      "events_gcs_sink.pipe",
+      `NODE export
+SQL >
+    SELECT * FROM events
+TYPE sink
+EXPORT_SERVICE gcs_hmac
+EXPORT_CONNECTION_NAME exports_gcs_hmac
+EXPORT_BUCKET_URI gcs://exports/events/
+EXPORT_FILE_TEMPLATE events_{date}
+EXPORT_SCHEDULE @once
+EXPORT_FORMAT ndjson
+EXPORT_STRATEGY create_new
+EXPORT_COMPRESSION gzip
+`
+    );
+
+    const result = await runMigrate({
+      cwd: tempDir,
+      patterns: ["."],
+      strict: true,
+    });
+
+    expect(result.success).toBe(true);
+    expect(result.errors).toHaveLength(0);
+
+    const output = fs.readFileSync(result.outputPath, "utf-8");
+    expect(output).toContain("defineGCSHmacConnection");
+    expect(output).toContain(
+      'export const exportsGcsHmac = defineGCSHmacConnection("exports_gcs_hmac", {'
+    );
+    expect(output).toContain('export const eventsGcsSink = defineSinkPipe("events_gcs_sink", {');
+    expect(output).toContain("sink: {");
+    expect(output).toContain("connection: exportsGcsHmac");
+    expect(output).toContain('bucketUri: "gcs://exports/events/"');
+    expect(output).toContain('fileTemplate: "events_{date}"');
+    expect(output).toContain('schedule: "@once"');
+    expect(output).toContain('format: "ndjson"');
+    expect(output).toContain('strategy: "create_new"');
+    expect(output).toContain('compression: "gzip"');
+  });
+
+  it("reports an error when a gcs_hmac sink references an s3 connection", async () => {
+    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "tinybird-migrate-"));
+    tempDirs.push(tempDir);
+
+    writeFile(
+      tempDir,
+      "exports_s3.connection",
+      `TYPE s3
+S3_REGION "us-east-1"
+S3_ARN "arn:aws:iam::123456789012:role/tinybird-s3-access"
+`
+    );
+
+    writeFile(
+      tempDir,
+      "events_gcs_sink.pipe",
+      `NODE export
+SQL >
+    SELECT * FROM events
+TYPE sink
+EXPORT_SERVICE gcs_hmac
+EXPORT_CONNECTION_NAME exports_s3
+EXPORT_BUCKET_URI gcs://exports/events/
+EXPORT_FILE_TEMPLATE events_{date}
+EXPORT_SCHEDULE @once
+EXPORT_FORMAT ndjson
+`
+    );
+
+    const result = await runMigrate({
+      cwd: tempDir,
+      patterns: ["."],
+      strict: true,
+    });
+
+    expect(result.success).toBe(false);
+    expect(result.errors.map((error) => error.message)).toContain(
+      'Sink pipe service "gcs_hmac" is incompatible with connection "exports_s3" type "s3".'
+    );
+  });
+
   it("migrates legacy EXPORT_WRITE_STRATEGY sink directives", async () => {
     const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "tinybird-migrate-"));
     tempDirs.push(tempDir);

src/generator/connection.test.ts

@@ -4,6 +4,7 @@ import {
   defineKafkaConnection,
   defineS3Connection,
   defineGCSConnection,
+  defineGCSHmacConnection,
   defineDynamoDBConnection,
 } from "../schema/connection.js";
 
@@ -220,6 +221,24 @@ describe("Connection Generator", () => {
       );
     });
 
+    it("generates GCS HMAC connection with access id and secret", () => {
+      const conn = defineGCSHmacConnection("my_gcs_hmac", {
+        accessId: '{{ tb_secret("GCS_ACCESS_ID") }}',
+        secret: '{{ tb_secret("GCS_SECRET") }}',
+      });
+
+      const result = generateConnection(conn);
+
+      expect(result.name).toBe("my_gcs_hmac");
+      expect(result.content).toBe(
+        [
+          "TYPE gcs_hmac",
+          'GCS_HMAC_ACCESS_ID {{ tb_secret("GCS_ACCESS_ID") }}',
+          'GCS_HMAC_SECRET {{ tb_secret("GCS_SECRET") }}',
+        ].join("\n")
+      );
+    });
+
     it("generates DynamoDB connection with arn and region", () => {
       const conn = defineDynamoDBConnection("my_dynamo", {
         region: "us-east-1",

src/generator/connection.ts

@@ -7,11 +7,13 @@ import type {
   ConnectionDefinition,
   KafkaConnectionDefinition,
   GCSConnectionDefinition,
+  GCSHmacConnectionDefinition,
   DynamoDBConnectionDefinition,
 } from "../schema/connection.js";
 import {
   isS3ConnectionDefinition,
   isGCSConnectionDefinition,
+  isGCSHmacConnectionDefinition,
   isDynamoDBConnectionDefinition,
   type S3ConnectionDefinition,
 } from "../schema/connection.js";
@@ -125,6 +127,20 @@ function generateGCSConnection(connection: GCSConnectionDefinition): string {
   return parts.join("\n");
 }
 
+/**
+ * Generate a GCS HMAC connection content
+ */
+function generateGCSHmacConnection(connection: GCSHmacConnectionDefinition): string {
+  const parts: string[] = [];
+  const options = connection.options;
+
+  parts.push("TYPE gcs_hmac");
+  parts.push(`GCS_HMAC_ACCESS_ID ${options.accessId}`);
+  parts.push(`GCS_HMAC_SECRET ${options.secret}`);
+
+  return parts.join("\n");
+}
+
 /**
  * Generate a DynamoDB connection content
  */
@@ -176,6 +192,8 @@ export function generateConnection(
     content = generateS3Connection(connection);
   } else if (isGCSConnectionDefinition(connection)) {
     content = generateGCSConnection(connection);
+  } else if (isGCSHmacConnectionDefinition(connection)) {
+    content = generateGCSHmacConnection(connection);
   } else if (isDynamoDBConnectionDefinition(connection)) {
     content = generateDynamoDBConnection(connection);
   } else {

src/generator/pipe.test.ts

@@ -2,7 +2,7 @@ import { describe, it, expect } from 'vitest';
 import { generatePipe, generateAllPipes } from './pipe.js';
 import { definePipe, defineEndpoint, defineMaterializedView, defineSinkPipe, node } from '../schema/pipe.js';
 import { defineDatasource } from '../schema/datasource.js';
-import { defineKafkaConnection, defineS3Connection } from '../schema/connection.js';
+import { defineKafkaConnection, defineS3Connection, defineGCSHmacConnection } from '../schema/connection.js';
 import { defineToken } from '../schema/token.js';
 import { t } from '../schema/types.js';
 import { p } from '../schema/params.js';
@@ -667,6 +667,36 @@ GROUP BY day, country
       expect(result.content).toContain('EXPORT_STRATEGY replace');
       expect(result.content).toContain('EXPORT_COMPRESSION gzip');
     });
+
+    it('generates GCS HMAC sink directives', () => {
+      const gcsHmac = defineGCSHmacConnection('exports_gcs_hmac', {
+        accessId: '{{ tb_secret("GCS_ACCESS_ID") }}',
+        secret: '{{ tb_secret("GCS_SECRET") }}',
+      });
+
+      const pipe = defineSinkPipe('events_gcs_sink', {
+        nodes: [node({ name: 'export', sql: 'SELECT * FROM events' })],
+        sink: {
+          connection: gcsHmac,
+          bucketUri: 'gcs://bucket/events/',
+          fileTemplate: 'events_{date}',
+          format: 'csv',
+          schedule: '@once',
+          compression: 'gzip',
+          strategy: 'replace',
+        },
+      });
+
+      const result = generatePipe(pipe);
+      expect(result.content).toContain('TYPE sink');
+      expect(result.content).toContain('EXPORT_CONNECTION_NAME exports_gcs_hmac');
+      expect(result.content).toContain('EXPORT_BUCKET_URI gcs://bucket/events/');
+      expect(result.content).toContain('EXPORT_FILE_TEMPLATE events_{date}');
+      expect(result.content).toContain('EXPORT_FORMAT csv');
+      expect(result.content).toContain('EXPORT_SCHEDULE @once');
+      expect(result.content).toContain('EXPORT_STRATEGY replace');
+      expect(result.content).toContain('EXPORT_COMPRESSION gzip');
+    });
   });
 
   describe('Token generation', () => {

src/index.ts

@@ -126,11 +126,13 @@ export {
   createKafkaConnection,
   defineS3Connection,
   defineGCSConnection,
+  defineGCSHmacConnection,
   defineDynamoDBConnection,
   isConnectionDefinition,
   isKafkaConnectionDefinition,
   isS3ConnectionDefinition,
   isGCSConnectionDefinition,
+  isGCSHmacConnectionDefinition,
   isDynamoDBConnectionDefinition,
   getConnectionType,
 } from "./schema/connection.js";
@@ -145,6 +147,8 @@ export type {
   S3ConnectionOptions,
   GCSConnectionDefinition,
   GCSConnectionOptions,
+  GCSHmacConnectionDefinition,
+  GCSHmacConnectionOptions,
   DynamoDBConnectionDefinition,
   DynamoDBConnectionOptions,
 } from "./schema/connection.js";
@@ -190,6 +194,7 @@ export type {
   SinkCompression,
   KafkaSinkConfig,
   S3SinkConfig,
+  GCSHmacSinkConfig,
   NodeDefinition,
   NodeOptions,
   ParamsDefinition,

src/migrate/emit-ts.ts

@@ -6,6 +6,7 @@ import type {
   DatasourceEngineModel,
   DynamoDBConnectionModel,
   GCSConnectionModel,
+  GCSHmacConnectionModel,
   KafkaConnectionModel,
   ParsedResource,
   PipeModel,
@@ -76,6 +77,9 @@ function hasSecretTemplate(resources: ParsedResource[]): boolean {
       } else if (resource.connectionType === "dynamodb") {
         values.push(resource.region);
         values.push(resource.arn);
+      } else if (resource.connectionType === "gcs_hmac") {
+        values.push(resource.accessId);
+        values.push(resource.secret);
       } else {
         values.push(resource.serviceAccountCredentialsJson);
       }
@@ -437,6 +441,7 @@ function emitConnection(
     | KafkaConnectionModel
     | S3ConnectionModel
     | GCSConnectionModel
+    | GCSHmacConnectionModel
     | DynamoDBConnectionModel
 ): string {
   const variableName = toCamelCase(connection.name);
@@ -518,6 +523,17 @@ function emitConnection(
     return lines.join("\n");
   }
 
+  if (connection.connectionType === "gcs_hmac") {
+    lines.push(
+      `export const ${variableName} = defineGCSHmacConnection(${escapeString(connection.name)}, {`
+    );
+    lines.push(`  accessId: ${emitStringOrSecret(connection.accessId)},`);
+    lines.push(`  secret: ${emitStringOrSecret(connection.secret)},`);
+    lines.push("});");
+    lines.push("");
+    return lines.join("\n");
+  }
+
   lines.push(
     `export const ${variableName} = defineGCSConnection(${escapeString(connection.name)}, {`
   );
@@ -663,6 +679,7 @@ export function emitMigrationFileContent(resources: ParsedResource[]): string {
       | KafkaConnectionModel
       | S3ConnectionModel
       | GCSConnectionModel
+      | GCSHmacConnectionModel
       | DynamoDBConnectionModel => resource.kind === "connection"
   );
   const datasources = resources.filter(
@@ -692,6 +709,9 @@ export function emitMigrationFileContent(resources: ParsedResource[]): string {
   if (connections.some((connection) => connection.connectionType === "gcs")) {
     imports.add("defineGCSConnection");
   }
+  if (connections.some((connection) => connection.connectionType === "gcs_hmac")) {
+    imports.add("defineGCSHmacConnection");
+  }
   if (connections.some((connection) => connection.connectionType === "dynamodb")) {
     imports.add("defineDynamoDBConnection");
   }

src/migrate/parse-connection.test.ts

@@ -163,4 +163,57 @@ S3_ARN arn:aws:iam::1:role/x`
       )
     ).toThrow("Kafka/S3/GCS directives are not valid for dynamodb connections.");
   });
+
+  it("parses basic GCS HMAC connection", () => {
+    const result = parseConnectionFile(
+      resource(
+        "my_gcs_hmac",
+        `TYPE gcs_hmac
+GCS_HMAC_ACCESS_ID {{ tb_secret("GCS_ACCESS_ID") }}
+GCS_HMAC_SECRET {{ tb_secret("GCS_SECRET") }}`
+      )
+    );
+
+    expect(result.connectionType).toBe("gcs_hmac");
+    expect(result).toHaveProperty("accessId", '{{ tb_secret("GCS_ACCESS_ID") }}');
+    expect(result).toHaveProperty("secret", '{{ tb_secret("GCS_SECRET") }}');
+  });
+
+  it("throws when GCS HMAC connection is missing GCS_HMAC_ACCESS_ID", () => {
+    expect(() =>
+      parseConnectionFile(
+        resource(
+          "my_gcs_hmac",
+          `TYPE gcs_hmac
+GCS_HMAC_SECRET {{ tb_secret("GCS_SECRET") }}`
+        )
+      )
+    ).toThrow("GCS_HMAC_ACCESS_ID is required");
+  });
+
+  it("throws when GCS HMAC connection is missing GCS_HMAC_SECRET", () => {
+    expect(() =>
+      parseConnectionFile(
+        resource(
+          "my_gcs_hmac",
+          `TYPE gcs_hmac
+GCS_HMAC_ACCESS_ID {{ tb_secret("GCS_ACCESS_ID") }}`
+        )
+      )
+    ).toThrow("GCS_HMAC_SECRET is required");
+  });
+
+  it("throws when GCS HMAC connection mixes GCS service account directives", () => {
+    expect(() =>
+      parseConnectionFile(
+        resource(
+          "my_gcs_hmac",
+          `TYPE gcs_hmac
+GCS_HMAC_ACCESS_ID {{ tb_secret("GCS_ACCESS_ID") }}
+GCS_HMAC_SECRET {{ tb_secret("GCS_SECRET") }}
+GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON {{ tb_secret("GCS_JSON") }}`
+        )
+      )
+    ).toThrow("Kafka/S3/GCS/DynamoDB directives are not valid for gcs_hmac connections.");
+  });
 });