Release v1.8.0

Author: exAspArk

Dockerfile

@@ -49,19 +49,25 @@ COPY --chown=app:app src/syncer-attio/lib/go.mod src/syncer-attio/lib/go.sum /ap
 COPY --chown=app:app src/syncer-attio/go.mod src/syncer-attio/go.sum /app/src/syncer-attio/
 RUN cd /app/src/syncer-attio && go mod download
 
+COPY --chown=app:app src/syncer-dialpad/lib/go.mod src/syncer-dialpad/lib/go.sum /app/src/syncer-dialpad/lib/
+COPY --chown=app:app src/syncer-dialpad/go.mod src/syncer-dialpad/go.sum /app/src/syncer-dialpad/
+RUN cd /app/src/syncer-dialpad && go mod download
+
 COPY --chown=app:app src/server/go.mod src/server/go.sum /app/src/server/
 RUN cd /app/src/server && go mod download
 
 COPY --chown=app:app src/common /app/src/common
 COPY --chown=app:app src/syncer-postgres /app/src/syncer-postgres
 COPY --chown=app:app src/syncer-amplitude /app/src/syncer-amplitude
 COPY --chown=app:app src/syncer-attio /app/src/syncer-attio
+COPY --chown=app:app src/syncer-dialpad /app/src/syncer-dialpad
 COPY --chown=app:app src/server /app/src/server
 
 RUN ARCH=$(dpkg --print-architecture) \
   && cd /app/src/syncer-postgres && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/syncer-postgres \
     && cd /app/src/syncer-amplitude && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/syncer-amplitude \
     && cd /app/src/syncer-attio && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/syncer-attio \
+    && cd /app/src/syncer-dialpad && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/syncer-dialpad \
     && cd /app/src/server && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/server
 
 # Prepare final image ##############################################################################
@@ -72,6 +78,7 @@ COPY --chown=app:app --from=compile \
   /app/bin/syncer-postgres \
   /app/bin/syncer-amplitude \
   /app/bin/syncer-attio \
+  /app/bin/syncer-dialpad \
   /app/bin/server \
   /app/bin/
 COPY --chown=app:app docker/bin /app/bin/

Makefile

@@ -6,13 +6,15 @@ install:
 		cd ../syncer-postgres && go mod tidy && cd ../lib && go mod tidy && \
 		cd ../../syncer-amplitude && go mod tidy && cd ./lib && go mod tidy && \
 		cd ../../syncer-attio && go mod tidy && cd ./lib && go mod tidy && \
+		cd ../../syncer-dialpad && go mod tidy && cd ./lib && go mod tidy && \
 		cd ../../server && go mod tidy"
 
 lint:
 	devbox run "cd src/common && go fmt && staticcheck . && \
 		cd ../syncer-postgres && go fmt && deadcode . && staticcheck . && cd ./lib && go fmt && staticcheck . && \
 		cd ../../syncer-amplitude && go fmt && deadcode . && staticcheck . && cd ./lib && go fmt && staticcheck . && \
 		cd ../../syncer-attio && go fmt && deadcode . && staticcheck . && cd ./lib && go fmt && staticcheck . && \
+		cd ../../syncer-dialpad && go fmt && deadcode . && staticcheck . && cd ./lib && go fmt && staticcheck . && \
 		cd ../../server && go fmt && deadcode . && staticcheck ."
 
 build:

README.md

@@ -10,6 +10,11 @@ BemiDB is an open-source Snowflake and Fivetran alternative bundled together. It
 - [Use cases](#use-cases)
 - [Quickstart](#quickstart)
 - [Usage](#usage)
+  - [Syncing from Amplitude](#syncing-from-amplitude)
+  - [Syncing from Attio](#syncing-from-attio)
+  - [Syncing from Dialpad](#syncing-from-dialpad)
+  - [Syncing from Postgres](#syncing-from-postgres)
+  - [Customizing S3 endpoint](#customizing-s3-endpoint)
 - [Configuration](#configuration)
 - [Architecture](#architecture)
 - [Benchmark](#benchmark)
@@ -118,50 +123,124 @@ psql postgres://localhost:54321/bemidb -c "SELECT COUNT(*) FROM postgres.[table_
 
 ## Usage
 
-#### Syncing from Postgres
+#### Syncing from Amplitude
 
-By default, BemiDB syncs all tables from the Postgres database. To include and sync only specific tables from your Postgres database:
+1. Create an [Amplitude API key](https://docs.gettelio.com/integrations/amplitude)
+2. Run the syncer:
 
 ```sh
 docker run \
-  -e SOURCE_POSTGRES_DATABASE_URL=postgres://user:password@host.docker.internal:5432/source \
-  -e SOURCE_POSTGRES_INCLUDE_TABLES=public.table1,public.table2 \ # A comma-separated list of tables to include
-  -e DESTINATION_SCHEMA_NAME=postgres \
+  -e SOURCE_AMPLITUDE_API_KEY=[...] \
+  -e SOURCE_AMPLITUDE_SECRET_KEY=[...] \
+  -e SOURCE_AMPLITUDE_START_DATE=2025-01-01 \
+  -e DESTINATION_SCHEMA_NAME=amplitude \
   -e AWS_REGION -e AWS_S3_BUCKET -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY -e CATALOG_DATABASE_URL \
-  ghcr.io/bemihq/bemidb:latest syncer-postgres
+  ghcr.io/bemihq/bemidb:latest syncer-amplitude
 ```
 
-To exclude specific tables during the sync:
+#### Syncing from Attio
+
+1. Create an [Attio API access token](https://docs.gettelio.com/integrations/attio)
+2. Run the syncer:
 
 ```sh
 docker run \
-  -e SOURCE_POSTGRES_DATABASE_URL=postgres://user:password@host.docker.internal:5432/source \
-  -e SOURCE_POSTGRES_EXCLUDE_TABLES=public.audit_log,public.cache \ # A comma-separated list of tables to exclude
-  -e DESTINATION_SCHEMA_NAME=postgres \
+  -e SOURCE_ATTIO_API_ACCESS_TOKEN=[...] \
+  -e DESTINATION_SCHEMA_NAME=attio \
   -e AWS_REGION -e AWS_S3_BUCKET -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY -e CATALOG_DATABASE_URL \
-  ghcr.io/bemihq/bemidb:latest syncer-postgres
+  ghcr.io/bemihq/bemidb:latest syncer-attio
 ```
 
-#### Syncing from Amplitude
+#### Syncing from Dialpad
+
+1. Create a [Dialpad API key](https://docs.gettelio.com/integrations/dialpad)
+2. Create a webhook endpoint:
+
+```sh
+curl -X POST "https://dialpad.com/api/v2/webhooks" \
+     -H "Content-Type: application/json" \
+     -H "Authorization: Bearer [DIALPAD_API_KEY]" \
+     -d '{
+           "hook_url": "https://[YOUR_DOMAIN]/[YOUR_WEBHOOK_ENDPOINT]",
+           "secret": "[YOUR_WEBHOOK_SECRET]"
+         }'
+```
+
+3. Subscribe to SMS events for the created webhook:
+
+```sh
+curl -X POST "https://dialpad.com/api/v2/subscriptions/sms" \
+     -H "Content-Type: application/json" \
+     -H "Authorization: Bearer [DIALPAD_API_KEY]" \
+     -d '{
+           "direction": "all",
+           "enabled": true,
+           "endpoint_id": "[WEBHOOK_ID]",
+           "include_internal": false,
+           "status": false
+         }'
+```
+
+4. Write a small service to receive Dialpad webhook events and publish them to NATS JetStream.
+
+<details>
+<summary>See example code in Node.js</summary>
+
+```ts
+import express from 'express';
+import bodyParser from 'body-parser';
+import { jwtVerify } from 'jose';
+import { connect, JSONCodec } from 'nats';
+
+const app = express();
+app.use(bodyParser.json());
+app.post('/dialpad-webhook', async (req, res) => {
+  const { payload } = await jwtVerify(request.body, new TextEncoder().encode('[YOUR_WEBHOOK_SECRET]'), { algorithms: ['HS256'] });
+  const jsonCodec = JSONCodec();
+  const natsConnection = await connect({ servers: "nats://host.docker.internal:4222" });
+  const jetstreamManager = await natsConnection.jetstreamManager();
+  await jetstreamManager.streams.add({ name: 'bemidb', subjects: ['bemidb.dialpad'] });
+  await jetstreamManager.jetstream().publish('bemidb.dialpad', jsonCodec.encode(payload));
+});
+app.listen(3000, () => console.log('Server is running on port 3000'));
+```
+</details>
+
+5. Run the syncer:
 
 ```sh
 docker run \
-  -e SOURCE_AMPLITUDE_API_KEY=[...] \
-  -e SOURCE_AMPLITUDE_SECRET_KEY=[...] \
-  -e SOURCE_AMPLITUDE_START_DATE=2025-01-01 \
-  -e DESTINATION_SCHEMA_NAME=amplitude \
+  -e NATS_URL=nats://host.docker.internal:4222 \
+  -e NATS_JETSTREAM_STREAM=bemidb \
+  -e NATS_JETSTREAM_SUBJECT=bemidb.dialpad \
+  -e NATS_JETSTREAM_CONSUMER_NAME=bemidb-dialpad \
+  -e DESTINATION_SCHEMA_NAME=dialpad \
   -e AWS_REGION -e AWS_S3_BUCKET -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY -e CATALOG_DATABASE_URL \
-  ghcr.io/bemihq/bemidb:latest syncer-amplitude
+  ghcr.io/bemihq/bemidb:latest syncer-dialpad
 ```
 
-#### Syncing from Attio
+#### Syncing from Postgres
+
+By default, BemiDB syncs all tables from the Postgres database. To include and sync only specific tables from your Postgres database:
 
 ```sh
 docker run \
-  -e SOURCE_ATTIO_API_ACCESS_TOKEN=[...] \
-  -e DESTINATION_SCHEMA_NAME=attio \
+  -e SOURCE_POSTGRES_DATABASE_URL=postgres://user:password@host.docker.internal:5432/source \
+  -e SOURCE_POSTGRES_INCLUDE_TABLES=public.table1,public.table2 \ # A comma-separated list of tables to include
+  -e DESTINATION_SCHEMA_NAME=postgres \
   -e AWS_REGION -e AWS_S3_BUCKET -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY -e CATALOG_DATABASE_URL \
-  ghcr.io/bemihq/bemidb:latest syncer-attio
+  ghcr.io/bemihq/bemidb:latest syncer-postgres
+```
+
+To exclude specific tables during the sync:
+
+```sh
+docker run \
+  -e SOURCE_POSTGRES_DATABASE_URL=postgres://user:password@host.docker.internal:5432/source \
+  -e SOURCE_POSTGRES_EXCLUDE_TABLES=public.audit_log,public.cache \ # A comma-separated list of tables to exclude
+  -e DESTINATION_SCHEMA_NAME=postgres \
+  -e AWS_REGION -e AWS_S3_BUCKET -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY -e CATALOG_DATABASE_URL \
+  ghcr.io/bemihq/bemidb:latest syncer-postgres
 ```
 
 #### Customizing S3 endpoint
@@ -195,15 +274,6 @@ export AWS_S3_ENDPOINT=http://localhost:9000
 
 ## Configuration
 
-#### `syncer-postgres` command options
-
-| Environment variable                | Default value | Description                                                          |
-|-------------------------------------|---------------|----------------------------------------------------------------------|
-| `DESTINATION_SCHEMA_NAME`           | Required      | Schema name in BemiDB to sync data to.                               |
-| `SOURCE_POSTGRES_DATABASE_URL`      | Required      | Postgres database URL to sync data from.                             |
-| `SOURCE_POSTGRES_INCLUDE_TABLES`    |               | List of tables to include in sync. Comma-separated `schema.table`.   |
-| `SOURCE_POSTGRES_EXCLUDE_TABLES`    |               | List of tables to exclude from sync. Comma-separated `schema.table`. |
-
 #### `syncer-amplitude` command options
 
 | Environment variable          | Default value | Description                                                        |
@@ -220,6 +290,26 @@ export AWS_S3_ENDPOINT=http://localhost:9000
 | `DESTINATION_SCHEMA_NAME`       | Required      | Schema name in BemiDB to sync data to.     |
 | `SOURCE_ATTIO_API_ACCESS_TOKEN` | Required      | Attio API access token for authentication. |
 
+#### `syncer-dialpad` command options
+
+| Environment variable           | Default value | Description                                                    |
+|--------------------------------|---------------|----------------------------------------------------------------|
+| `DESTINATION_SCHEMA_NAME`      | Required      | Schema name in BemiDB to sync data to.                         |
+| `NATS_URL`                     | Required      | NATS server URL for connecting to receive Dialpad SMS records. |
+| `NATS_JETSTREAM_STREAM`        | Required      | NATS JetStream stream name.                                    |
+| `NATS_JETSTREAM_SUBJECT`       | Required      | NATS JetStream subject name.                                   |
+| `NATS_JETSTREAM_CONSUMER_NAME` | Required      | NATS JetStream consumer name.                                  |
+| `NATS_FETCH_TIMEOUT_SECONDS`   | `30`          | Timeout in seconds for fetching messages from NATS.            |
+
+#### `syncer-postgres` command options
+
+| Environment variable                | Default value | Description                                                          |
+|-------------------------------------|---------------|----------------------------------------------------------------------|
+| `DESTINATION_SCHEMA_NAME`           | Required      | Schema name in BemiDB to sync data to.                               |
+| `SOURCE_POSTGRES_DATABASE_URL`      | Required      | Postgres database URL to sync data from.                             |
+| `SOURCE_POSTGRES_INCLUDE_TABLES`    |               | List of tables to include in sync. Comma-separated `schema.table`.   |
+| `SOURCE_POSTGRES_EXCLUDE_TABLES`    |               | List of tables to exclude from sync. Comma-separated `schema.table`. |
+
 #### `server` command options
 
 | Environment variable | Default value | Description                            |
@@ -323,6 +413,7 @@ SELECT * FROM [TABLE] WHERE [JSON_COLUMN]->>'[JSON_KEY]' = '[JSON_VALUE]';
   - [x] Amplitude (incremental)
   - [x] Attio CRM (full-refresh)
   - [x] Postgres (full-refresh)
+  - [x] Dialpad (real-time)
   - [ ] HubSpot
   - [ ] Stripe
   - [ ] Google Sheets

docker/bin/run.sh

@@ -37,6 +37,19 @@ case "${1:-}" in
     ./bin/syncer-attio 2>&1 | sed 's/^/[Syncer] /'
     echo "Syncer for Attio finished."
     ;;
+  syncer-dialpad)
+    : "${NATS_URL:?Environment variable NATS_URL must be set}"
+    : "${NATS_JETSTREAM_STREAM:?Environment variable NATS_JETSTREAM_STREAM must be set}"
+    : "${NATS_JETSTREAM_SUBJECT:?Environment variable NATS_JETSTREAM_SUBJECT must be set}"
+    : "${NATS_JETSTREAM_CONSUMER_NAME:?Environment variable NATS_JETSTREAM_CONSUMER_NAME must be set}"
+    : "${DESTINATION_SCHEMA_NAME:?Environment variable DESTINATION_SCHEMA_NAME must be set}"
+
+    psql $CATALOG_DATABASE_URL -f /app/scripts/catalog.sql
+
+    echo "Starting Syncer for Dialpad..."
+    ./bin/syncer-dialpad 2>&1 | sed 's/^/[Syncer] /'
+    echo "Syncer for Dialpad finished."
+    ;;
   server)
     : "${AWS_REGION:?Environment variable AWS_REGION must be set}"
     : "${AWS_S3_BUCKET:?Environment variable AWS_S3_BUCKET must be set}"

src/common/duckdb_client.go

@@ -14,52 +14,46 @@ var SYNCER_DUCKDB_BOOT_QUERIES = []string{
 }
 
 type DuckdbClient struct {
-	Config    *CommonConfig
-	Db        *sql.DB
-	Connector *duckdb.Connector
+	Config      *CommonConfig
+	Db          *sql.DB
+	Connector   *duckdb.Connector
+	BootQueries []string
 }
 
 func NewDuckdbClient(config *CommonConfig, bootQueries ...[]string) *DuckdbClient {
 	ctx := context.Background()
 	connector, err := duckdb.NewConnector("", nil)
 	PanicIfError(config, err)
 	db := sql.OpenDB(connector)
-	PanicIfError(config, err)
 
 	client := &DuckdbClient{
 		Config:    config,
 		Db:        db,
 		Connector: connector,
 	}
 
-	queries := []string{
+	client.BootQueries = []string{
 		"SET timezone='UTC'",
 	}
 	if bootQueries != nil {
-		queries = append(queries, bootQueries[0]...)
-	}
-	for _, query := range queries {
-		_, err := client.ExecContext(ctx, query)
-		PanicIfError(config, err)
+		client.BootQueries = append(client.BootQueries, bootQueries[0]...)
 	}
-
-	client.setExplicitAwsCredentials(ctx)
-
+	client.BootQueries = append(
+		client.BootQueries,
+		"CREATE OR REPLACE SECRET aws_s3_secret (TYPE S3, KEY_ID '"+config.Aws.AccessKeyId+"', SECRET '"+config.Aws.SecretAccessKey+"', REGION '"+config.Aws.Region+"', ENDPOINT '"+config.Aws.S3Endpoint+"', SCOPE 's3://"+config.Aws.S3Bucket+"')",
+	)
 	if IsLocalHost(config.Aws.S3Endpoint) {
-		_, err = client.ExecContext(ctx, "SET s3_use_ssl=false")
-		PanicIfError(config, err)
+		client.BootQueries = append(client.BootQueries, "SET s3_use_ssl=false")
 	}
-
 	if config.Aws.S3Endpoint != DEFAULT_AWS_S3_ENDPOINT {
-		// Use endpoint/bucket/key (path, deprecated on AWS) instead of bucket.endpoint/key (vhost)
-		_, err = client.ExecContext(ctx, "SET s3_url_style='path'")
-		PanicIfError(config, err)
+		client.BootQueries = append(client.BootQueries, "SET s3_url_style='path'") // Use endpoint/bucket/key (path, deprecated on AWS) instead of bucket.endpoint/key (vhost)
 	}
-
 	if config.LogLevel == LOG_LEVEL_TRACE {
-		_, err = client.ExecContext(ctx, "PRAGMA enable_logging('HTTP')")
-		PanicIfError(config, err)
-		_, err = client.ExecContext(ctx, "SET logging_storage = 'stdout'")
+		client.BootQueries = append(client.BootQueries, "PRAGMA enable_logging('HTTP')", "SET logging_storage = 'stdout'")
+	}
+
+	for _, query := range client.BootQueries {
+		_, err := client.ExecContext(ctx, query)
 		PanicIfError(config, err)
 	}
 
@@ -130,17 +124,22 @@ func (client *DuckdbClient) Close() {
 	client.Db.Close()
 }
 
-func (client *DuckdbClient) setExplicitAwsCredentials(ctx context.Context) {
-	config := client.Config
-	query := "CREATE OR REPLACE SECRET aws_s3_secret (TYPE S3, KEY_ID '$accessKeyId', SECRET '$secretAccessKey', REGION '$region', ENDPOINT '$endpoint', SCOPE '$s3Bucket')"
-	_, err := client.ExecContext(ctx, query, map[string]string{
-		"accessKeyId":     config.Aws.AccessKeyId,
-		"secretAccessKey": config.Aws.SecretAccessKey,
-		"region":          config.Aws.Region,
-		"endpoint":        config.Aws.S3Endpoint,
-		"s3Bucket":        "s3://" + config.Aws.S3Bucket,
-	})
-	PanicIfError(config, err)
+func (client *DuckdbClient) RecreateDb() {
+	ctx := context.Background()
+
+	client.Db.Close()
+
+	connector, err := duckdb.NewConnector("", nil)
+	PanicIfError(client.Config, err)
+	db := sql.OpenDB(connector)
+	client.Db = db
+	client.Connector = connector
+
+	for _, query := range client.BootQueries {
+		_, err := client.Db.ExecContext(ctx, query)
+		PanicIfError(client.Config, err)
+	}
+
 }
 
 func replaceNamedStringArgs(query string, args map[string]string) string {