diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 29a2841b25..4630ba2858 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -1266,6 +1266,9 @@ importers:
       '@crowd/redis':
         specifier: workspace:*
         version: link:../../libs/redis
+      '@crowd/snowflake':
+        specifier: workspace:*
+        version: link:../../libs/snowflake
       '@crowd/types':
         specifier: workspace:*
         version: link:../../libs/types
diff --git a/services/apps/script_executor_worker/package.json b/services/apps/script_executor_worker/package.json
index c634aad1e3..c020c48e42 100644
--- a/services/apps/script_executor_worker/package.json
+++ b/services/apps/script_executor_worker/package.json
@@ -6,6 +6,7 @@
     "start:debug": "CROWD_TEMPORAL_TASKQUEUE=script-executor SERVICE=script-executor-worker LOG_LEVEL=info tsx --inspect=0.0.0.0:9232 src/main.ts",
     "dev:local": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug:local",
     "dev": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug",
+    "cleanup-fork-activities": "npx tsx src/bin/cleanup-fork-activities-and-maintainers.ts",
     "lint": "npx eslint --ext .ts src --max-warnings=0",
     "format": "npx prettier --write \"src/**/*.ts\"",
     "format-check": "npx prettier --check .",
@@ -19,6 +20,7 @@
     "@crowd/logging": "workspace:*",
     "@crowd/opensearch": "workspace:*",
     "@crowd/redis": "workspace:*",
+    "@crowd/snowflake": "workspace:*",
     "@crowd/types": "workspace:*",
     "@temporalio/client": "~1.11.8",
     "@temporalio/workflow": "~1.11.8",
diff --git a/services/apps/script_executor_worker/src/bin/cleanup-fork-activities-and-maintainers.ts b/services/apps/script_executor_worker/src/bin/cleanup-fork-activities-and-maintainers.ts
new file mode 100644
index 0000000000..97d508f390
--- /dev/null
+++ b/services/apps/script_executor_worker/src/bin/cleanup-fork-activities-and-maintainers.ts
@@ -0,0 +1,605 @@
+/**
+ * Fork Activities and Maintainers Cleanup Script
+ *
+ * PROBLEM:
+ * When a fork repository is onboarded, all activities from the original project are wrongly
+ * attributed to the new, forked project. This makes the data inaccurate.
+ *
+ * SOLUTION:
+ * This script deletes all activities and maintainers from forked repositories across:
+ * - Tinybird
+ * - CDP Postgres
+ *
+ * Usage:
+ *   # Via package.json script (recommended):
+ *   pnpm run cleanup-fork-activities -- <repo-url> [<repo-url> ...] [--dry-run] [--tb-token <token>]
+ *
+ *   # Or directly with tsx:
+ *   npx tsx src/bin/cleanup-fork-activities-and-maintainers.ts <repo-url> [<repo-url> ...] [--dry-run] [--tb-token <token>]
+ *
+ * Options:
+ *   --dry-run          Display what would be deleted without actually deleting anything
+ *   --tb-token         Tinybird API token to use (overrides CROWD_TINYBIRD_ACTIVITIES_TOKEN environment variable)
+ *
+ * Environment Variables Required:
+ *   CROWD_DB_WRITE_HOST - Postgres write host
+ *   CROWD_DB_PORT - Postgres port
+ *   CROWD_DB_USERNAME - Postgres username
+ *   CROWD_DB_PASSWORD - Postgres password
+ *   CROWD_DB_DATABASE - Postgres database name
+ *   CROWD_TINYBIRD_BASE_URL - Tinybird API base URL
+ *   CROWD_TINYBIRD_ACTIVITIES_TOKEN - Tinybird API token
+ */
+import {
+  TinybirdClient,
+  WRITE_DB_CONFIG,
+  getDbConnection,
+} from '@crowd/data-access-layer/src/database'
+import { QueryExecutor, pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { getServiceChildLogger } from '@crowd/logging'
+
+const log = getServiceChildLogger('cleanup-fork-activities-script')
+
+// Type definitions
+interface ForkRepository {
+  id: string
+  url: string
+  segmentId: string
+  forkedFrom: string
+}
+
+interface DatabaseClients {
+  postgres: QueryExecutor
+}
+
+/**
+ * Initialize Postgres connection using QueryExecutor
+ */
+async function initPostgresClient(): Promise<QueryExecutor> {
+  log.info('Initializing Postgres connection...')
+
+  const dbConnection = await getDbConnection(WRITE_DB_CONFIG())
+  const queryExecutor = pgpQx(dbConnection)
+
+  log.info('Postgres connection established')
+  return queryExecutor
+}
+
+/**
+ * Initialize all database clients
+ */
+async function initDatabaseClients(): Promise<DatabaseClients> {
+  log.info('Initializing database clients...')
+
+  const postgres = await initPostgresClient()
+
+  log.info('All database clients initialized successfully')
+
+  return {
+    postgres,
+  }
+}
+
+/**
+ * Lookup a single fork repository by URL
+ */
+async function lookupForkRepository(postgres: QueryExecutor, url: string): Promise<ForkRepository> {
+  log.info(`Looking up repository by URL: ${url}`)
+
+  const query = `
+    SELECT 
+      id,
+      url,
+      "segmentId",
+      "forkedFrom"
+    FROM git.repositories
+    WHERE url = $(url)
+      AND "forkedFrom" IS NOT NULL
+  `
+
+  const repo = (await postgres.selectOneOrNone(query, { url })) as ForkRepository | null
+
+  if (!repo) {
+    // Check if repository exists but is not a fork
+    const checkQuery = `
+      SELECT id, url, "forkedFrom"
+      FROM git.repositories
+      WHERE url = $(url)
+    `
+    const check = await postgres.selectOneOrNone(checkQuery, { url })
+
+    if (!check) {
+      throw new Error(`Repository not found in database: ${url}`)
+    } else {
+      throw new Error(`Repository is not a fork (forkedFrom is null): ${url}`)
+    }
+  }
+
+  log.info(`✓ Found fork repository: ${repo.url} → ${repo.forkedFrom}`)
+  return repo
+}
+
+/**
+ * Lock a repository to prevent concurrent cleanup operations
+ */
+async function lockRepository(
+  postgres: QueryExecutor,
+  repoId: string,
+  repoUrl: string,
+  dryRun = false,
+): Promise<void> {
+  if (dryRun) {
+    log.info(`[DRY RUN] Would lock repository: ${repoUrl}`)
+    return
+  }
+
+  log.info(`Locking repository: ${repoUrl}`)
+
+  const query = `
+    UPDATE git.repositories
+    SET "lockedAt" = NOW()
+    WHERE id = $(repoId)
+  `
+
+  await postgres.result(query, { repoId })
+  log.info(`✓ Repository locked: ${repoUrl}`)
+}
+
+/**
+ * Unlock a repository after cleanup is complete
+ */
+async function unlockRepository(
+  postgres: QueryExecutor,
+  repoId: string,
+  repoUrl: string,
+  dryRun = false,
+): Promise<void> {
+  if (dryRun) {
+    log.info(`[DRY RUN] Would unlock repository: ${repoUrl}`)
+    return
+  }
+
+  log.info(`Unlocking repository: ${repoUrl}`)
+
+  const query = `
+    UPDATE git.repositories
+    SET "lockedAt" = NULL
+    WHERE id = $(repoId)
+  `
+
+  await postgres.result(query, { repoId })
+  log.info(`✓ Repository unlocked: ${repoUrl}`)
+}
+
+/**
+ * Delete maintainers for a fork repository from Postgres
+ */
+async function deleteMaintainersFromPostgres(
+  postgres: QueryExecutor,
+  repoId: string,
+  repoUrl: string,
+  dryRun = false,
+): Promise<number> {
+  if (dryRun) {
+    log.info(`[DRY RUN] Querying maintainers for repository: ${repoUrl}`)
+    const query = `
+      SELECT COUNT(*) as count
+      FROM "maintainersInternal"
+      WHERE "repoId" = $(repoId)
+    `
+    const result = (await postgres.selectOne(query, { repoId })) as { count: string }
+    const rowCount = parseInt(result.count, 10)
+    log.info(`[DRY RUN] Would delete ${rowCount} maintainer(s) from Postgres`)
+    return rowCount
+  }
+
+  log.info(`Deleting maintainers for repository: ${repoUrl}`)
+  const query = `
+    DELETE FROM "maintainersInternal"
+    WHERE "repoId" = $(repoId)
+  `
+  const rowCount = await postgres.result(query, { repoId })
+  log.info(`✓ Deleted ${rowCount} maintainer(s) from Postgres for repository ${repoUrl}`)
+  return rowCount
+}
+
+/**
+ * Delete maintainers from Tinybird using the delete API
+ * Note: Tinybird deletions are async and may take time to reflect
+ * See: https://www.tinybird.co/docs/classic/get-data-in/data-operations/replace-and-delete-data#delete-data-selectively
+ */
+async function deleteMaintainersFromTinybird(
+  tinybird: TinybirdClient,
+  repoId: string,
+  repoUrl: string,
+  dryRun = false,
+): Promise<number> {
+  if (dryRun) {
+    log.info(`[DRY RUN] Querying maintainers from Tinybird for repository: ${repoUrl}`)
+    try {
+      const query = `SELECT COUNT(*) as count FROM maintainersInternal WHERE repoId = '${repoId}' FORMAT JSON`
+      const result = await tinybird.executeSql<{ data: Array<{ count: string }> }>(query)
+      const count = result.data.length > 0 ? parseInt(result.data[0].count, 10) : 0
+      log.info(`[DRY RUN] Would delete ${count} maintainer(s) from Tinybird`)
+      return count
+    } catch (error) {
+      log.error(`Failed to query maintainers from Tinybird: ${error.message}`)
+      throw error
+    }
+  }
+
+  log.info(`Deleting maintainers from Tinybird for repository: ${repoUrl}`)
+
+  try {
+    // Delete from maintainersInternal datasource using repoId
+    log.info('Deleting from maintainersInternal datasource...')
+    const deleteCondition = `repoId = '${repoId}'`
+    const jobResponse = await tinybird.deleteDatasource('maintainersInternal', deleteCondition)
+
+    log.info(
+      `✓ Submitted deletion job to Tinybird (job_id: ${jobResponse.job_id}, job_url: ${jobResponse.job_url})`,
+    )
+    log.info(
+      `  Note: Deletions are async and may take time to complete. Check job status at: ${jobResponse.job_url}`,
+    )
+    return 0 // Tinybird deletion is async, so we can't return actual count
+  } catch (error) {
+    log.error(`Failed to delete maintainers from Tinybird: ${error.message}`)
+    throw error
+  }
+}
+
+/**
+ * Query activity IDs from Tinybird for a fork repository
+ * This must be done before deletion because Tinybird deletions are asynchronous
+ */
+async function queryActivityIds(
+  tinybird: TinybirdClient,
+  segmentId: string,
+  channel: string,
+): Promise<string[]> {
+  log.info(`Querying activity IDs from Tinybird for segment: ${segmentId}, channel: ${channel}`)
+
+  try {
+    const query = `SELECT activityId FROM activityRelations WHERE segmentId = '${segmentId}' AND channel = '${channel}' AND platform = 'git' FORMAT JSON`
+    const result = await tinybird.executeSql<{ data: Array<{ activityId: string }> }>(query)
+
+    const activityIds = result.data.map((row) => row.activityId)
+    log.info(`Found ${activityIds.length} activity ID(s) in Tinybird`)
+    return activityIds
+  } catch (error) {
+    log.error(`Failed to query activity IDs from Tinybird: ${error.message}`)
+    throw error
+  }
+}
+
+/**
+ * Delete activities from Tinybird using the delete API
+ * Note: Tinybird deletions are async and may take time to reflect
+ * See: https://www.tinybird.co/docs/classic/get-data-in/data-operations/replace-and-delete-data#delete-data-selectively
+ */
+async function deleteActivitiesFromTinybird(
+  tinybird: TinybirdClient,
+  activityIds: string[],
+  dryRun = false,
+): Promise<void> {
+  if (activityIds.length === 0) {
+    log.info('No activities to delete from Tinybird')
+    return
+  }
+
+  if (dryRun) {
+    log.info(`[DRY RUN] Would delete ${activityIds.length} activities from Tinybird`)
+    log.info(
+      `[DRY RUN] Would delete from 'activities' datasource: ${activityIds.length} activity(ies)`,
+    )
+    log.info(
+      `[DRY RUN] Would delete from 'activityRelations' datasource: ${activityIds.length} relation(s)`,
+    )
+    return
+  }
+
+  log.info(`Deleting ${activityIds.length} activities from Tinybird...`)
+
+  // Format activity IDs for SQL IN clause
+  const idsString = activityIds.map((id) => `'${id}'`).join(',')
+
+  try {
+    // Delete from activities datasource
+    log.info('Deleting from activities datasource...')
+    const activitiesDeleteCondition = `id IN (${idsString})`
+    const activitiesJobResponse = await tinybird.deleteDatasource(
+      'activities',
+      activitiesDeleteCondition,
+    )
+    log.info(`✓ Submitted deletion job for activities (job_id: ${activitiesJobResponse.job_id})`)
+
+    // Delete from activityRelations datasource
+    log.info('Deleting from activityRelations datasource...')
+    const activityRelationsDeleteCondition = `activityId IN (${idsString})`
+    const activityRelationsJobResponse = await tinybird.deleteDatasource(
+      'activityRelations',
+      activityRelationsDeleteCondition,
+    )
+    log.info(
+      `✓ Submitted deletion job for activityRelations (job_id: ${activityRelationsJobResponse.job_id})`,
+    )
+
+    log.info(
+      `✓ Submitted deletion jobs to Tinybird (note: deletions are async and may take time to complete)`,
+    )
+    log.info(`  Activities job URL: ${activitiesJobResponse.job_url}`)
+    log.info(`  ActivityRelations job URL: ${activityRelationsJobResponse.job_url}`)
+  } catch (error) {
+    log.error(`Failed to delete activities from Tinybird: ${error.message}`)
+    throw error
+  }
+}
+
+/**
+ * Delete activity relations from Postgres
+ */
+async function deleteActivityRelationsFromPostgres(
+  postgres: QueryExecutor,
+  activityIds: string[],
+  dryRun = false,
+): Promise<number> {
+  if (activityIds.length === 0) {
+    log.info(`No activity IDs to ${dryRun ? 'query' : 'delete'} from Postgres`)
+    return 0
+  }
+
+  if (dryRun) {
+    log.info(`[DRY RUN] Querying ${activityIds.length} activity relations from Postgres...`)
+    const query = `
+      SELECT COUNT(*) as count
+      FROM "activityRelations"
+      WHERE "activityId" IN ($(activityIds:csv))
+    `
+    const result = (await postgres.selectOne(query, { activityIds })) as { count: string }
+    const rowCount = parseInt(result.count, 10)
+    log.info(`[DRY RUN] Would delete ${rowCount} activity relation(s) from Postgres`)
+    return rowCount
+  }
+
+  log.info(`Deleting ${activityIds.length} activity relations from Postgres...`)
+  const query = `
+    DELETE FROM "activityRelations"
+    WHERE "activityId" IN ($(activityIds:csv))
+  `
+  const rowCount = await postgres.result(query, { activityIds })
+  log.info(`✓ Deleted ${rowCount} activity relation(s) from Postgres`)
+  return rowCount
+}
+
+/**
+ * Process cleanup for a single fork repository
+ */
+async function cleanupForkRepository(
+  clients: DatabaseClients,
+  repo: ForkRepository,
+  dryRun = false,
+  tbToken?: string,
+): Promise<void> {
+  if (dryRun) {
+    log.info(`\n${'='.repeat(80)}`)
+    log.info(`[DRY RUN MODE] Processing: ${repo.url}`)
+    log.info(`${'='.repeat(80)}`)
+  } else {
+    log.info(`\n${'='.repeat(80)}`)
+    log.info(`Processing: ${repo.url}`)
+    log.info(`${'='.repeat(80)}`)
+  }
+
+  // Lock repository to prevent concurrent operations
+  await lockRepository(clients.postgres, repo.id, repo.url, dryRun)
+
+  try {
+    // Initialize Tinybird client once for this repository
+    const tinybird = new TinybirdClient(tbToken)
+
+    // Step 1: Delete maintainers from all systems
+    log.info('Deleting maintainers from all systems...')
+    const maintainersDeletedPostgres = await deleteMaintainersFromPostgres(
+      clients.postgres,
+      repo.id,
+      repo.url,
+      dryRun,
+    )
+    const maintainersDeletedTinybird = await deleteMaintainersFromTinybird(
+      tinybird,
+      repo.id,
+      repo.url,
+      dryRun,
+    )
+
+    // Step 2: Query activity IDs from Tinybird
+    const activityIds = await queryActivityIds(tinybird, repo.segmentId, repo.url)
+    log.info(`Found ${activityIds.length} activity ID(s) to ${dryRun ? 'query' : 'delete'}`)
+
+    if (activityIds.length === 0) {
+      log.info('No activities to delete, skipping deletion steps')
+      log.info(`✓ Completed ${dryRun ? 'dry run for' : 'cleanup for'} ${repo.url}`)
+      log.info(
+        `  - Maintainers ${dryRun ? 'found' : 'deleted'} (Postgres): ${maintainersDeletedPostgres}`,
+      )
+      log.info(
+        `  - Maintainers ${dryRun ? 'found' : 'deleted'} (Tinybird): ${maintainersDeletedTinybird}`,
+      )
+      return
+    }
+
+    // Process activities in batches of 200
+    const BATCH_SIZE = 200
+    const batches: string[][] = []
+    for (let i = 0; i < activityIds.length; i += BATCH_SIZE) {
+      batches.push(activityIds.slice(i, i + BATCH_SIZE))
+    }
+
+    log.info(
+      `Processing ${activityIds.length} activities in ${batches.length} batch(es) of up to ${BATCH_SIZE}`,
+    )
+
+    let totalActivityRelationsDeleted = 0
+
+    for (let batchIndex = 0; batchIndex < batches.length; batchIndex++) {
+      const batch = batches[batchIndex]
+      log.info(
+        `Processing batch ${batchIndex + 1}/${batches.length} (${batch.length} activities)...`,
+      )
+
+      // Step 3: Delete from Postgres
+      const activityRelationsDeleted = await deleteActivityRelationsFromPostgres(
+        clients.postgres,
+        batch,
+        dryRun,
+      )
+      totalActivityRelationsDeleted += activityRelationsDeleted
+
+      // Step 4: Delete from Tinybird last (source of truth - delete last so we can retry if needed)
+      await deleteActivitiesFromTinybird(tinybird, batch, dryRun)
+
+      log.info(`✓ Completed batch ${batchIndex + 1}/${batches.length}`)
+    }
+
+    log.info(`✓ Completed ${dryRun ? 'dry run for' : 'cleanup for'} ${repo.url}`)
+    log.info(
+      `  - Maintainers ${dryRun ? 'found' : 'deleted'} (Postgres): ${maintainersDeletedPostgres}`,
+    )
+    log.info(
+      `  - Maintainers ${dryRun ? 'found' : 'deleted'} (Tinybird): ${maintainersDeletedTinybird}`,
+    )
+    log.info(`  - Activities ${dryRun ? 'found' : 'deleted'}: ${activityIds.length}`)
+    log.info(
+      `  - Activity relations ${dryRun ? 'found' : 'deleted'}: ${totalActivityRelationsDeleted}`,
+    )
+  } catch (error) {
+    log.error(`Failed to cleanup repository ${repo.url}: ${error.message}`)
+    throw error
+  } finally {
+    // Always unlock repository, even if cleanup failed
+    await unlockRepository(clients.postgres, repo.id, repo.url, dryRun)
+  }
+}
+
+/**
+ * Main entry point
+ */
+async function main() {
+  const args = process.argv.slice(2)
+
+  // Parse flags
+  const dryRunIndex = args.indexOf('--dry-run')
+  const tbTokenIndex = args.indexOf('--tb-token')
+  const dryRun = dryRunIndex !== -1
+
+  // Extract tb-token value if provided
+  let tbToken: string | undefined
+  if (tbTokenIndex !== -1) {
+    if (tbTokenIndex + 1 >= args.length) {
+      log.error('Error: --tb-token requires a value')
+      process.exit(1)
+    }
+    tbToken = args[tbTokenIndex + 1]
+  }
+
+  // Remove flags and their values from args to get URLs
+  const urls = args.filter(
+    (arg, index) =>
+      index !== dryRunIndex &&
+      index !== tbTokenIndex &&
+      (tbTokenIndex === -1 || index !== tbTokenIndex + 1),
+  )
+
+  if (urls.length === 0) {
+    log.error(`
+      Usage:
+        # Via package.json script (recommended):
+        pnpm run cleanup-fork-activities -- <repo-url> [<repo-url> ...] [--dry-run] [--tb-token <token>]
+        
+        # Or directly with tsx:
+        npx tsx src/bin/cleanup-fork-activities-and-maintainers.ts <repo-url> [<repo-url> ...] [--dry-run] [--tb-token <token>]
+      
+      Arguments:
+        repo-url: One or more repository URLs to clean up
+        --dry-run: (optional) Display what would be deleted without actually deleting anything
+        --tb-token: (optional) Tinybird API token to use (overrides CROWD_TINYBIRD_ACTIVITIES_TOKEN environment variable)
+      
+      Examples:
+        # Clean up a single repository
+        pnpm run cleanup-fork-activities -- https://github.com/owner/repo1
+        
+        # Clean up multiple repositories
+        pnpm run cleanup-fork-activities -- https://github.com/owner/repo1 https://github.com/owner/repo2
+        
+        # Dry run to preview what would be deleted
+        pnpm run cleanup-fork-activities -- https://github.com/owner/repo1 --dry-run
+        
+        # Use custom Tinybird token
+        pnpm run cleanup-fork-activities -- https://github.com/owner/repo1 --tb-token your-token-here
+      
+      Note: 
+        - URLs must exist in git.repositories table
+        - Repository must be a fork (forkedFrom field must not be null)
+    `)
+    process.exit(1)
+  }
+
+  if (dryRun) {
+    log.info(`\n${'='.repeat(80)}`)
+    log.info('🧪 DRY RUN MODE - No data will be deleted')
+    log.info(`${'='.repeat(80)}\n`)
+  }
+
+  try {
+    log.info(`Processing ${urls.length} repository URL(s)`)
+
+    // Initialize database clients
+    const clients = await initDatabaseClients()
+
+    // Process cleanup workflow
+    log.info(`\n${'='.repeat(80)}`)
+    log.info(`Starting ${dryRun ? 'dry run (preview)' : 'cleanup'} workflow`)
+    log.info(`${'='.repeat(80)}`)
+
+    let successCount = 0
+    let failureCount = 0
+    const failedRepos: string[] = []
+
+    for (const url of urls) {
+      try {
+        // Lookup and cleanup in the same loop
+        const repo = await lookupForkRepository(clients.postgres, url)
+        await cleanupForkRepository(clients, repo, dryRun, tbToken)
+        successCount++
+      } catch (error) {
+        failureCount++
+        failedRepos.push(url)
+        log.error(`Failed to process ${url}: ${error.message}`)
+        // Continue with next repository instead of stopping
+      }
+    }
+
+    // Summary
+    log.info(`\n${'='.repeat(80)}`)
+    log.info('Cleanup Summary')
+    log.info(`${'='.repeat(80)}`)
+    log.info(`✓ Successfully processed: ${successCount}/${urls.length}`)
+    if (failureCount > 0) {
+      log.warn(`✗ Failed: ${failureCount}/${urls.length}`)
+      log.warn('Failed repositories:')
+      failedRepos.forEach((url) => log.warn(`  - ${url}`))
+    }
+
+    process.exit(failureCount > 0 ? 1 : 0)
+  } catch (error) {
+    log.error(error, 'Failed to run cleanup script')
+    log.error(`\n❌ Error: ${error.message}`)
+    process.exit(1)
+  }
+}
+
+main().catch((error) => {
+  log.error('Unexpected error:', error)
+  process.exit(1)
+})
diff --git a/services/libs/database/src/tinybirdClient.ts b/services/libs/database/src/tinybirdClient.ts
index 575274c024..028707b05a 100644
--- a/services/libs/database/src/tinybirdClient.ts
+++ b/services/libs/database/src/tinybirdClient.ts
@@ -42,9 +42,27 @@ export class TinybirdClient {
     keepAlive: false, // Disable keep-alive to avoid stale socket reuse
   })
 
-  constructor() {
+  constructor(token?: string) {
     this.host = process.env.CROWD_TINYBIRD_BASE_URL ?? 'https://api.tinybird.co'
-    this.token = process.env.CROWD_TINYBIRD_ACTIVITIES_TOKEN ?? ''
+    this.token = token ?? process.env.CROWD_TINYBIRD_ACTIVITIES_TOKEN ?? ''
+  }
+
+  /**
+   * Get common headers for Tinybird API requests
+   * @param contentType - Optional Content-Type header value (default: undefined)
+   * @returns Headers object for axios requests
+   */
+  private getHeaders(contentType?: string): Record<string, string> {
+    const headers: Record<string, string> = {
+      Authorization: `Bearer ${this.token}`,
+      Accept: 'application/json',
+    }
+
+    if (contentType) {
+      headers['Content-Type'] = contentType
+    }
+
+    return headers
   }
 
   /**
@@ -69,10 +87,7 @@ export class TinybirdClient {
     }`
 
     const result = await axios.get<T>(url, {
-      headers: {
-        Authorization: `Bearer ${this.token}`,
-        Accept: 'application/json',
-      },
+      headers: this.getHeaders(),
       httpsAgent: TinybirdClient.httpsAgent,
     })
 
@@ -92,21 +107,18 @@ export class TinybirdClient {
       }
     }
 
-    // Compose the final request body
-    const url = `${this.host}/v0/sql`
-    const body: Record<string, unknown> = {
-      q: `% SELECT * FROM ${pipeName} FORMAT JSON`,
-      format: 'json',
-    }
+    const query = `% SELECT * FROM ${pipeName} FORMAT JSON`
 
     // Copy user params as-is, preserving arrays and primitives
+    const bodyParams: Record<string, unknown> = { format: 'json' }
     for (const [k, v] of Object.entries(params)) {
       if (v === undefined || v === null) continue
+
       // Sanity: Tinybird accepts arrays, booleans, numbers, strings
       if (Array.isArray(v)) {
-        body[k] = v.map((x) => String(x))
+        bodyParams[k] = v.map((x) => String(x))
       } else if (typeof v === 'string' || typeof v === 'number' || typeof v === 'boolean') {
-        body[k] = v
+        bodyParams[k] = v
       } else {
         // If you accidentally pass a Date here, throw to avoid silent mismatch
         throw new Error(
@@ -115,12 +127,61 @@ export class TinybirdClient {
       }
     }
 
+    return await this.executeSql<T>(query, bodyParams)
+  }
+
+  /**
+   * Execute raw SQL query on Tinybird
+   * Useful for queries that don't go through named pipes (e.g., ALTER TABLE, direct SELECT)
+   */
+  async executeSql<T = unknown>(query: string, bodyParams?: Record<string, unknown>): Promise<T> {
+    const url = `${this.host}/v0/sql`
+    const body: Record<string, unknown> = { q: query, ...bodyParams }
+
     const result = await axios.post<T>(url, body, {
-      headers: {
-        Authorization: `Bearer ${this.token}`,
-        'Content-Type': 'application/json',
-        Accept: 'application/json',
-      },
+      headers: this.getHeaders('application/json'),
+      responseType: 'json',
+      httpsAgent: TinybirdClient.httpsAgent,
+    })
+
+    return result.data
+  }
+
+  /**
+   * Delete data from a Tinybird datasource using the delete API
+   * See: https://www.tinybird.co/docs/classic/get-data-in/data-operations/replace-and-delete-data#delete-data-selectively
+   *
+   * @param datasourceName - Name of the datasource to delete from
+   * @param deleteCondition - SQL expression filter (e.g., "repoId = 'xxx'", "id IN ('a', 'b')")
+   * @returns Job response with job_id and job_url for tracking deletion progress
+   */
+  async deleteDatasource(
+    datasourceName: string,
+    deleteCondition: string,
+  ): Promise<{
+    id: string
+    job_id: string
+    job_url: string
+    status: string
+    job: {
+      kind: string
+      id: string
+      job_id: string
+      status: string
+      datasource: {
+        id: string
+        name: string
+      }
+      delete_condition: string
+    }
+  }> {
+    const url = `${this.host}/v0/datasources/${encodeURIComponent(datasourceName)}/delete`
+
+    // Tinybird expects URL-encoded form data, not JSON
+    const payload = `delete_condition=${encodeURIComponent(deleteCondition)}`
+
+    const result = await axios.post(url, payload, {
+      headers: this.getHeaders('application/x-www-form-urlencoded'),
       responseType: 'json',
       httpsAgent: TinybirdClient.httpsAgent,
     })