From 66c299fc24aacb9ecd7fc2ccf5010bd724b7cc07 Mon Sep 17 00:00:00 2001 From: Tommy Healy Date: Thu, 1 Oct 2026 17:20:46 +0200 Subject: [PATCH] feat: add GCS HMAC connector distinction and GCS sink destination Add defineGCSHmacConnection for GCS's S3-compatible HMAC (access-key-id + secret) auth mode, distinct from defineGCSConnection's service account JSON mode, and extend defineSinkPipe's sink destinations with a GCS HMAC bucket export variant alongside the existing Kafka and S3 ones. - schema/connection.ts: GCSHmacConnectionOptions/Definition, defineGCSHmacConnection, isGCSHmacConnectionDefinition - generator/connection.ts: emits `TYPE gcs_hmac` / GCS_HMAC_ACCESS_ID / GCS_HMAC_SECRET - schema/pipe.ts: GCSHmacSinkConfig (same shape as S3SinkConfig), validateSinkConfig accepts S3 or GCS HMAC connections for bucket sinks - migrate/*: parse-connection.ts, parse-pipe.ts, emit-ts.ts, and types.ts extended so the datafile->TS migration path round-trips gcs_hmac connections and sinks instead of rejecting/misclassifying them (src/codegen/ has no connection/sink reverse-codegen at all today, so nothing to extend there) - index.ts: exports the new public types/functions Field names (TYPE gcs_hmac, GCS_HMAC_ACCESS_ID, GCS_HMAC_SECRET, EXPORT_SERVICE gcs_hmac) were traced to tinybird/connector_settings.py, tinybird/tb/modules/connection.py, and tinybird/tb/modules/datafile/format_connection.py in the analytics monorepo (DataConnectorType.GCLOUD_STORAGE_HMAC = "gcs_hmac"). Per tinybird/datafile/common.py, the modern Forward backend treats EXPORT_SERVICE as deprecated/inferred-from-connection for sink pipes, so the TS generator (like the existing S3 sink) does not emit it. Refs PROTM-2269 Co-Authored-By: Claude Sonnet 5 --- src/cli/commands/migrate.test.ts | 97 ++++++++++++++++++++++++++++ src/generator/connection.test.ts | 19 ++++++ src/generator/connection.ts | 18 ++++++ src/generator/pipe.test.ts | 32 ++++++++- src/index.ts | 5 ++ src/migrate/emit-ts.ts | 20 ++++++ src/migrate/parse-connection.test.ts | 53 +++++++++++++++ src/migrate/parse-connection.ts | 86 +++++++++++++++++++++++- src/migrate/parse-pipe.ts | 62 +++++++++++------- src/migrate/types.ts | 23 ++++++- src/schema/connection.test.ts | 76 ++++++++++++++++++++++ src/schema/connection.ts | 82 +++++++++++++++++++++++ src/schema/pipe.test.ts | 47 +++++++++++++- src/schema/pipe.ts | 43 ++++++++++-- 14 files changed, 630 insertions(+), 33 deletions(-) diff --git a/src/cli/commands/migrate.test.ts b/src/cli/commands/migrate.test.ts index 0bef5b0..bededd3 100644 --- a/src/cli/commands/migrate.test.ts +++ b/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); diff --git a/src/generator/connection.test.ts b/src/generator/connection.test.ts index fff6d6f..3ce21ec 100644 --- a/src/generator/connection.test.ts +++ b/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", diff --git a/src/generator/connection.ts b/src/generator/connection.ts index a52ebd0..58c2fa4 100644 --- a/src/generator/connection.ts +++ b/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 { diff --git a/src/generator/pipe.test.ts b/src/generator/pipe.test.ts index 391caaa..6af7cef 100644 --- a/src/generator/pipe.test.ts +++ b/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', () => { diff --git a/src/index.ts b/src/index.ts index b8bab18..e05b946 100644 --- a/src/index.ts +++ b/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, diff --git a/src/migrate/emit-ts.ts b/src/migrate/emit-ts.ts index 473eb10..a0e5e8a 100644 --- a/src/migrate/emit-ts.ts +++ b/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"); } diff --git a/src/migrate/parse-connection.test.ts b/src/migrate/parse-connection.test.ts index 0521f6a..d72adf8 100644 --- a/src/migrate/parse-connection.test.ts +++ b/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."); + }); }); diff --git a/src/migrate/parse-connection.ts b/src/migrate/parse-connection.ts index 39f4a10..02d2a16 100644 --- a/src/migrate/parse-connection.ts +++ b/src/migrate/parse-connection.ts @@ -1,6 +1,7 @@ import type { DynamoDBConnectionModel, GCSConnectionModel, + GCSHmacConnectionModel, KafkaConnectionModel, ResourceFile, S3ConnectionModel, @@ -32,6 +33,8 @@ const CONNECTION_DIRECTIVES = new Set([ "S3_ACCESS_KEY", "S3_SECRET", "GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON", + "GCS_HMAC_ACCESS_ID", + "GCS_HMAC_SECRET", "DYNAMODB_ARN", "DYNAMODB_REGION", ]); @@ -46,7 +49,12 @@ function isConnectionDirectiveLine(line: string): boolean { export function parseConnectionFile( resource: ResourceFile -): KafkaConnectionModel | S3ConnectionModel | GCSConnectionModel | DynamoDBConnectionModel { +): + | KafkaConnectionModel + | S3ConnectionModel + | GCSConnectionModel + | GCSHmacConnectionModel + | DynamoDBConnectionModel { const lines = splitLines(resource.content); let connectionType: string | undefined; @@ -77,6 +85,9 @@ export function parseConnectionFile( let accessSecret: string | undefined; let serviceAccountCredentialsJson: string | undefined; + let gcsHmacAccessId: string | undefined; + let gcsHmacSecret: string | undefined; + let dynamodbArn: string | undefined; let dynamodbRegion: string | undefined; @@ -183,6 +194,12 @@ export function parseConnectionFile( } serviceAccountCredentialsJson = parseQuotedValue(value); break; + case "GCS_HMAC_ACCESS_ID": + gcsHmacAccessId = parseQuotedValue(value); + break; + case "GCS_HMAC_SECRET": + gcsHmacSecret = parseQuotedValue(value); + break; case "DYNAMODB_ARN": dynamodbArn = parseQuotedValue(value); break; @@ -216,6 +233,8 @@ export function parseConnectionFile( accessKey || accessSecret || serviceAccountCredentialsJson || + gcsHmacAccessId || + gcsHmacSecret || dynamodbArn || dynamodbRegion ) { @@ -269,6 +288,8 @@ export function parseConnectionFile( schemaRegistryUrl || sslCaPem || serviceAccountCredentialsJson || + gcsHmacAccessId || + gcsHmacSecret || dynamodbArn || dynamodbRegion ) { @@ -336,6 +357,8 @@ export function parseConnectionFile( arn || accessKey || accessSecret || + gcsHmacAccessId || + gcsHmacSecret || dynamodbArn || dynamodbRegion ) { @@ -365,6 +388,63 @@ export function parseConnectionFile( }; } + if (connectionType === "gcs_hmac") { + if ( + bootstrapServers || + securityProtocol || + saslMechanism || + saslOauthbearerMethod || + saslOauthbearerAwsRegion || + saslOauthbearerAwsRoleArn || + saslOauthbearerAwsExternalId || + key || + secret || + schemaRegistryUrl || + sslCaPem || + region || + arn || + accessKey || + accessSecret || + serviceAccountCredentialsJson || + dynamodbArn || + dynamodbRegion + ) { + throw new MigrationParseError( + resource.filePath, + "connection", + resource.name, + "Kafka/S3/GCS/DynamoDB directives are not valid for gcs_hmac connections." + ); + } + + if (!gcsHmacAccessId) { + throw new MigrationParseError( + resource.filePath, + "connection", + resource.name, + "GCS_HMAC_ACCESS_ID is required for gcs_hmac connections." + ); + } + + if (!gcsHmacSecret) { + throw new MigrationParseError( + resource.filePath, + "connection", + resource.name, + "GCS_HMAC_SECRET is required for gcs_hmac connections." + ); + } + + return { + kind: "connection", + name: resource.name, + filePath: resource.filePath, + connectionType: "gcs_hmac", + accessId: gcsHmacAccessId, + secret: gcsHmacSecret, + }; + } + if (connectionType === "dynamodb") { if ( bootstrapServers || @@ -382,7 +462,9 @@ export function parseConnectionFile( arn || accessKey || accessSecret || - serviceAccountCredentialsJson + serviceAccountCredentialsJson || + gcsHmacAccessId || + gcsHmacSecret ) { throw new MigrationParseError( resource.filePath, diff --git a/src/migrate/parse-pipe.ts b/src/migrate/parse-pipe.ts index 9438de7..edaa27d 100644 --- a/src/migrate/parse-pipe.ts +++ b/src/migrate/parse-pipe.ts @@ -492,7 +492,7 @@ export function parsePipeFile(resource: ResourceFile): PipeModel { let copyTargetDatasource: string | undefined; let copySchedule: string | undefined; let copyMode: "append" | "replace" | undefined; - let exportService: "kafka" | "s3" | undefined; + let exportService: "kafka" | "s3" | "gcs_hmac" | undefined; let exportConnectionName: string | undefined; let exportTopic: string | undefined; let exportBucketUri: string | undefined; @@ -662,7 +662,7 @@ export function parsePipeFile(resource: ResourceFile): PipeModel { break; case "EXPORT_SERVICE": { const normalized = parseQuotedValue(value).toLowerCase(); - if (normalized !== "kafka" && normalized !== "s3") { + if (normalized !== "kafka" && normalized !== "s3" && normalized !== "gcs_hmac") { throw new MigrationParseError( resource.filePath, "pipe", @@ -813,23 +813,28 @@ export function parsePipeFile(resource: ResourceFile): PipeModel { } const hasKafkaDirectives = exportTopic !== undefined; - const hasS3Directives = + const hasBucketDirectives = exportBucketUri !== undefined || exportFileTemplate !== undefined || exportFormat !== undefined || exportCompression !== undefined; - if (hasKafkaDirectives && hasS3Directives) { + if (hasKafkaDirectives && hasBucketDirectives) { throw new MigrationParseError( resource.filePath, "pipe", resource.name, - "Sink pipe cannot mix Kafka and S3 export directives." + "Sink pipe cannot mix Kafka and S3/GCS HMAC export directives." ); } + // Bucket-based sinks (S3 and GCS HMAC) share the exact same EXPORT_* directive + // shape, so when EXPORT_SERVICE is omitted (it's deprecated/inferred from the + // connection by the Forward backend) we can't tell them apart from directives + // alone. Default to "s3" in that case, matching pre-existing behavior; an + // explicit EXPORT_SERVICE gcs_hmac is required to migrate a GCS HMAC sink. const inferredService = - exportService ?? (hasKafkaDirectives ? "kafka" : hasS3Directives ? "s3" : undefined); + exportService ?? (hasKafkaDirectives ? "kafka" : hasBucketDirectives ? "s3" : undefined); if (!inferredService) { throw new MigrationParseError( @@ -841,12 +846,12 @@ export function parsePipeFile(resource: ResourceFile): PipeModel { } if (inferredService === "kafka") { - if (hasS3Directives) { + if (hasBucketDirectives) { throw new MigrationParseError( resource.filePath, "pipe", resource.name, - "S3 export directives are not valid for Kafka sinks." + "S3/GCS HMAC export directives are not valid for Kafka sinks." ); } if (!exportTopic) { @@ -870,7 +875,7 @@ export function parsePipeFile(resource: ResourceFile): PipeModel { resource.filePath, "pipe", resource.name, - "EXPORT_STRATEGY is only valid for S3 sinks." + "EXPORT_STRATEGY is only valid for S3/GCS HMAC sinks." ); } if (exportCompression !== undefined) { @@ -878,7 +883,7 @@ export function parsePipeFile(resource: ResourceFile): PipeModel { resource.filePath, "pipe", resource.name, - "EXPORT_COMPRESSION is only valid for S3 sinks." + "EXPORT_COMPRESSION is only valid for S3/GCS HMAC sinks." ); } @@ -894,7 +899,7 @@ export function parsePipeFile(resource: ResourceFile): PipeModel { resource.filePath, "pipe", resource.name, - "Kafka export directives are not valid for S3 sinks." + "Kafka export directives are not valid for S3/GCS HMAC sinks." ); } if (!exportBucketUri || !exportFileTemplate || !exportFormat || !exportSchedule) { @@ -902,20 +907,33 @@ export function parsePipeFile(resource: ResourceFile): PipeModel { resource.filePath, "pipe", resource.name, - "S3 sinks require EXPORT_BUCKET_URI, EXPORT_FILE_TEMPLATE, EXPORT_FORMAT, and EXPORT_SCHEDULE." + "S3/GCS HMAC sinks require EXPORT_BUCKET_URI, EXPORT_FILE_TEMPLATE, EXPORT_FORMAT, and EXPORT_SCHEDULE." ); } - sink = { - service: "s3", - connectionName: exportConnectionName, - bucketUri: exportBucketUri, - fileTemplate: exportFileTemplate, - format: exportFormat, - schedule: exportSchedule, - strategy: exportStrategy, - compression: exportCompression, - }; + if (inferredService === "gcs_hmac") { + sink = { + service: "gcs_hmac", + connectionName: exportConnectionName, + bucketUri: exportBucketUri, + fileTemplate: exportFileTemplate, + format: exportFormat, + schedule: exportSchedule, + strategy: exportStrategy, + compression: exportCompression, + }; + } else { + sink = { + service: "s3", + connectionName: exportConnectionName, + bucketUri: exportBucketUri, + fileTemplate: exportFileTemplate, + format: exportFormat, + schedule: exportSchedule, + strategy: exportStrategy, + compression: exportCompression, + }; + } } } diff --git a/src/migrate/types.ts b/src/migrate/types.ts index 05b48e1..3a3d34b 100644 --- a/src/migrate/types.ts +++ b/src/migrate/types.ts @@ -125,7 +125,18 @@ export interface PipeS3SinkModel { compression?: "none" | "gzip" | "snappy"; } -export type PipeSinkModel = PipeKafkaSinkModel | PipeS3SinkModel; +export interface PipeGCSHmacSinkModel { + service: "gcs_hmac"; + connectionName: string; + bucketUri: string; + fileTemplate: string; + format: string; + schedule: string; + strategy?: "create_new" | "replace"; + compression?: "none" | "gzip" | "snappy"; +} + +export type PipeSinkModel = PipeKafkaSinkModel | PipeS3SinkModel | PipeGCSHmacSinkModel; export interface PipeParamModel { name: string; @@ -191,6 +202,15 @@ export interface GCSConnectionModel { serviceAccountCredentialsJson: string; } +export interface GCSHmacConnectionModel { + kind: "connection"; + name: string; + filePath: string; + connectionType: "gcs_hmac"; + accessId: string; + secret: string; +} + export interface DynamoDBConnectionModel { kind: "connection"; name: string; @@ -206,6 +226,7 @@ export type ParsedResource = | KafkaConnectionModel | S3ConnectionModel | GCSConnectionModel + | GCSHmacConnectionModel | DynamoDBConnectionModel; export interface MigrationResult { diff --git a/src/schema/connection.test.ts b/src/schema/connection.test.ts index 2c0afe9..242a507 100644 --- a/src/schema/connection.test.ts +++ b/src/schema/connection.test.ts @@ -3,11 +3,13 @@ import { defineKafkaConnection, defineS3Connection, defineGCSConnection, + defineGCSHmacConnection, defineDynamoDBConnection, isConnectionDefinition, isKafkaConnectionDefinition, isS3ConnectionDefinition, isGCSConnectionDefinition, + isGCSHmacConnectionDefinition, isDynamoDBConnectionDefinition, getConnectionType, } from "./connection.js"; @@ -201,6 +203,48 @@ describe("Connection Schema", () => { }); }); + describe("defineGCSHmacConnection", () => { + it("creates a GCS HMAC connection with required fields", () => { + const conn = defineGCSHmacConnection("my_gcs_hmac", { + accessId: '{{ tb_secret("GCS_ACCESS_ID") }}', + secret: '{{ tb_secret("GCS_SECRET") }}', + }); + + expect(conn._name).toBe("my_gcs_hmac"); + expect(conn._type).toBe("connection"); + expect(conn._connectionType).toBe("gcs_hmac"); + expect(conn.options.accessId).toBe('{{ tb_secret("GCS_ACCESS_ID") }}'); + expect(conn.options.secret).toBe('{{ tb_secret("GCS_SECRET") }}'); + }); + + it("throws when accessId is missing", () => { + expect(() => + defineGCSHmacConnection("my_gcs_hmac", { + accessId: " ", + secret: '{{ tb_secret("GCS_SECRET") }}', + }) + ).toThrow("GCS HMAC connection `accessId` is required."); + }); + + it("throws when secret is missing", () => { + expect(() => + defineGCSHmacConnection("my_gcs_hmac", { + accessId: '{{ tb_secret("GCS_ACCESS_ID") }}', + secret: "", + }) + ).toThrow("GCS HMAC connection `secret` is required."); + }); + + it("throws for invalid connection name", () => { + expect(() => + defineGCSHmacConnection("123-invalid", { + accessId: '{{ tb_secret("GCS_ACCESS_ID") }}', + secret: '{{ tb_secret("GCS_SECRET") }}', + }) + ).toThrow("Invalid connection name"); + }); + }); + describe("defineDynamoDBConnection", () => { it("creates a DynamoDB connection with required fields", () => { const conn = defineDynamoDBConnection("my_dynamo", { @@ -308,6 +352,29 @@ describe("Connection Schema", () => { }); }); + describe("isGCSHmacConnectionDefinition", () => { + it("returns true for GCS HMAC connection", () => { + const conn = defineGCSHmacConnection("my_gcs_hmac", { + accessId: '{{ tb_secret("GCS_ACCESS_ID") }}', + secret: '{{ tb_secret("GCS_SECRET") }}', + }); + + expect(isGCSHmacConnectionDefinition(conn)).toBe(true); + }); + + it("returns false for non-GCS-HMAC objects", () => { + expect(isGCSHmacConnectionDefinition({})).toBe(false); + expect(isGCSHmacConnectionDefinition(null)).toBe(false); + expect( + isGCSHmacConnectionDefinition( + defineGCSConnection("my_gcs", { + serviceAccountCredentialsJson: '{{ tb_secret("GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON") }}', + }) + ) + ).toBe(false); + }); + }); + describe("isDynamoDBConnectionDefinition", () => { it("returns true for DynamoDB connection", () => { const conn = defineDynamoDBConnection("my_dynamo", { @@ -354,5 +421,14 @@ describe("Connection Schema", () => { expect(getConnectionType(conn)).toBe("gcs"); }); + + it("returns the gcs_hmac connection type", () => { + const conn = defineGCSHmacConnection("my_gcs_hmac", { + accessId: '{{ tb_secret("GCS_ACCESS_ID") }}', + secret: '{{ tb_secret("GCS_SECRET") }}', + }); + + expect(getConnectionType(conn)).toBe("gcs_hmac"); + }); }); }); diff --git a/src/schema/connection.ts b/src/schema/connection.ts index e3dac3d..556004f 100644 --- a/src/schema/connection.ts +++ b/src/schema/connection.ts @@ -117,6 +117,34 @@ export interface GCSConnectionDefinition { readonly options: GCSConnectionOptions; } +/** + * Options for creating a GCS HMAC connection. + * + * HMAC keys are an S3-compatible, access-key-id/secret auth mode for GCS, + * as an alternative to uploading a service account credentials JSON. + */ +export interface GCSHmacConnectionOptions { + /** GCS HMAC access key id - can use {{ tb_secret(...) }} */ + accessId: string; + /** GCS HMAC secret - can use {{ tb_secret(...) }} */ + secret: string; +} + +/** + * GCS HMAC-specific connection definition + */ +export interface GCSHmacConnectionDefinition { + readonly [CONNECTION_BRAND]: true; + /** Connection name */ + readonly _name: string; + /** Type marker for inference */ + readonly _type: "connection"; + /** Connection type */ + readonly _connectionType: "gcs_hmac"; + /** GCS HMAC options */ + readonly options: GCSHmacConnectionOptions; +} + /** * Options for defining a DynamoDB connection */ @@ -149,6 +177,7 @@ export type ConnectionDefinition = | KafkaConnectionDefinition | S3ConnectionDefinition | GCSConnectionDefinition + | GCSHmacConnectionDefinition | DynamoDBConnectionDefinition; function validateConnectionName(name: string): void { @@ -257,6 +286,50 @@ export function defineGCSConnection( }; } +/** + * Define a GCS HMAC connection + * + * HMAC connections authenticate to GCS using an S3-compatible access key id + * and secret, as an alternative to `defineGCSConnection`'s service account + * credentials JSON. + * + * @param name - The connection name (must be valid identifier) + * @param options - GCS HMAC connection configuration + * @returns A connection definition that can be used in a project + * + * @example + * ```ts + * import { defineGCSHmacConnection } from '@tinybirdco/sdk'; + * + * export const myGcsHmac = defineGCSHmacConnection('my_gcs_hmac', { + * accessId: '{{ tb_secret("GCS_ACCESS_ID") }}', + * secret: '{{ tb_secret("GCS_SECRET") }}', + * }); + * ``` + */ +export function defineGCSHmacConnection( + name: string, + options: GCSHmacConnectionOptions +): GCSHmacConnectionDefinition { + validateConnectionName(name); + + if (!options.accessId?.trim()) { + throw new Error("GCS HMAC connection `accessId` is required."); + } + + if (!options.secret?.trim()) { + throw new Error("GCS HMAC connection `secret` is required."); + } + + return { + [CONNECTION_BRAND]: true, + _name: name, + _type: "connection", + _connectionType: "gcs_hmac", + options, + }; +} + /** * Define a DynamoDB connection * @@ -330,6 +403,15 @@ export function isGCSConnectionDefinition(value: unknown): value is GCSConnectio return isConnectionDefinition(value) && value._connectionType === "gcs"; } +/** + * Check if a value is a GCS HMAC connection definition + */ +export function isGCSHmacConnectionDefinition( + value: unknown +): value is GCSHmacConnectionDefinition { + return isConnectionDefinition(value) && value._connectionType === "gcs_hmac"; +} + /** * Check if a value is a DynamoDB connection definition */ diff --git a/src/schema/pipe.test.ts b/src/schema/pipe.test.ts index 84a47fe..8bf4679 100644 --- a/src/schema/pipe.test.ts +++ b/src/schema/pipe.test.ts @@ -15,7 +15,7 @@ import { sql, } from "./pipe.js"; import { defineDatasource } from "./datasource.js"; -import { defineKafkaConnection, defineS3Connection } from "./connection.js"; +import { defineKafkaConnection, defineS3Connection, defineGCSHmacConnection } from "./connection.js"; import { t } from "./types.js"; import { p } from "./params.js"; import { engine } from "./engines.js"; @@ -229,6 +229,51 @@ describe("Pipe Schema", () => { expect(isSinkPipe(pipe)).toBe(true); }); + it("creates a GCS HMAC sink pipe", () => { + const gcsHmac = defineGCSHmacConnection("exports_gcs_hmac", { + accessId: '{{ tb_secret("GCS_ACCESS_ID") }}', + secret: '{{ tb_secret("GCS_SECRET") }}', + }); + + const pipe = defineSinkPipe("exports_gcs_sink", { + nodes: [node({ name: "export", sql: "SELECT * FROM events" })], + sink: { + connection: gcsHmac, + bucketUri: "gcs://exports/events/", + fileTemplate: "events_{date}", + schedule: "@once", + format: "csv", + strategy: "create_new", + compression: "gzip", + }, + }); + + const sink = getSinkConfig(pipe); + expect(sink).toBeTruthy(); + expect(sink && "bucketUri" in sink ? sink.bucketUri : undefined).toBe("gcs://exports/events/"); + expect(isSinkPipe(pipe)).toBe(true); + }); + + it("throws when bucket sink connection type is invalid", () => { + const kafka = defineKafkaConnection("events_kafka", { + bootstrapServers: "localhost:9092", + }); + + expect(() => + defineSinkPipe("bad_bucket_sink", { + nodes: [node({ name: "export", sql: "SELECT * FROM events" })], + sink: { + // Runtime validation rejects mismatched connection/type + connection: kafka as unknown as ReturnType, + bucketUri: "s3://exports/events/", + fileTemplate: "events_{date}", + schedule: "@once", + format: "csv", + }, + }) + ).toThrow("requires an S3 or GCS HMAC connection"); + }); + it("throws when Kafka sink connection type is invalid", () => { const s3 = defineS3Connection("exports_s3", { region: "us-east-1", diff --git a/src/schema/pipe.ts b/src/schema/pipe.ts index 69db0a5..1363d98 100644 --- a/src/schema/pipe.ts +++ b/src/schema/pipe.ts @@ -9,8 +9,16 @@ import type { DatasourceDefinition, SchemaDefinition, ColumnDefinition } from ". import { getColumnType } from "./datasource.js"; import { getTinybirdType } from "./types.js"; import type { TokenDefinition, PipeTokenScope } from "./token.js"; -import type { KafkaConnectionDefinition, S3ConnectionDefinition } from "./connection.js"; -import { isKafkaConnectionDefinition, isS3ConnectionDefinition } from "./connection.js"; +import type { + KafkaConnectionDefinition, + S3ConnectionDefinition, + GCSHmacConnectionDefinition, +} from "./connection.js"; +import { + isKafkaConnectionDefinition, + isS3ConnectionDefinition, + isGCSHmacConnectionDefinition, +} from "./connection.js"; /** Symbol for brand typing pipes - use Symbol.for() for global registry */ export const PIPE_BRAND = Symbol.for("tinybird.pipe"); @@ -202,9 +210,32 @@ export interface S3SinkConfig { } /** - * Sink pipe configuration (Kafka or S3 only) + * GCS HMAC sink configuration + * + * GCS HMAC export uses the same kind of access-key-id/secret-based + * connection and bucket export shape as S3. + */ +export interface GCSHmacSinkConfig { + /** GCS HMAC connection used to write exported files */ + connection: GCSHmacConnectionDefinition; + /** Destination bucket URI (for example: gcs://bucket/prefix/) */ + bucketUri: string; + /** Output filename template (supports Tinybird placeholders) */ + fileTemplate: string; + /** Output format (for example: csv, ndjson) */ + format: string; + /** Sink schedule (for example: @on-demand, @once, cron expression) */ + schedule: string; + /** Export strategy */ + strategy?: SinkStrategy; + /** Compression codec */ + compression?: SinkCompression; +} + +/** + * Sink pipe configuration (Kafka, S3, or GCS HMAC) */ -export type SinkConfig = KafkaSinkConfig | S3SinkConfig; +export type SinkConfig = KafkaSinkConfig | S3SinkConfig | GCSHmacSinkConfig; /** * Inline token configuration for pipe access @@ -522,9 +553,9 @@ function validateSinkConfig(pipeName: string, sink: SinkConfig): void { return; } - if (!isS3ConnectionDefinition(sink.connection)) { + if (!isS3ConnectionDefinition(sink.connection) && !isGCSHmacConnectionDefinition(sink.connection)) { throw new Error( - `Pipe "${pipeName}" S3 sink requires an S3 connection.` + `Pipe "${pipeName}" bucket sink requires an S3 or GCS HMAC connection.` ); } if (typeof sink.bucketUri !== "string" || !sink.bucketUri.trim()) {