[WIP, NO MERGE]: Added recurring job to sync automation runs to Tinybird

ghost/core/core/boot.js

@@ -473,93 +473,6 @@ async function initBackgroundServices({ config }) {
     return;
   }
 
-  const tinybirdConfig = config.get('tinybird:stats');
-  const tinybirdEndpoint = tinybirdConfig?.local?.enabled
-    ? tinybirdConfig.local.endpoint
-    : tinybirdConfig?.endpoint;
-
-  if (tinybirdEndpoint && Math.random() === 100) {
-    const db = require('./server/data/db');
-    const logging = require('@tryghost/logging');
-    const settingsCache = require('./shared/settings-cache');
-    const token = config.get('tinybird:adminToken');
-    const siteUuid = tinybirdConfig.id || settingsCache.get('site_uuid');
-    const batchSize = 10_000;
-
-    const syncAutomationEvents = async ({ table, datasource, name }) => {
-      try {
-        if (!token) {
-          logging.info(`Skipping ${name} sync to Tinybird: no admin token configured`);
-          return;
-        }
-
-        let lastId;
-        let totalSent = 0;
-
-        while (true) {
-          const query = db.knex(table).select('*').orderBy('id').limit(batchSize);
-          if (lastId) {
-            query.where('id', '>', lastId);
-          }
-
-          const rows = await query;
-          if (!rows.length) {
-            if (!totalSent) {
-              logging.info(`Skipping ${name} sync to Tinybird: no ${name}s found`);
-            }
-            return;
-          }
-
-          const events = rows.map((row) => ({
-            site_uuid: siteUuid,
-            id: row.id,
-            updated_at: row.updated_at,
-            payload: {
-              ...row,
-              site_uuid: siteUuid,
-            },
-          }));
-
-          logging.info(`Sending ${events.length} ${name} events to Tinybird`);
-
-          const response = await fetch(`${tinybirdEndpoint}/v0/events?name=${datasource}`, {
-            method: 'POST',
-            headers: {
-              Authorization: `Bearer ${token}`,
-              'Content-Type': 'application/x-ndjson',
-            },
-            body: events.map((event) => JSON.stringify(event)).join('\n'),
-          });
-
-          if (!response.ok) {
-            // I don't care
-            // eslint-disable-next-line
-            throw new Error(`Tinybird API error: ${response.status} - ${await response.text()}`);
-          }
-
-          totalSent += events.length;
-          lastId = rows.at(-1).id;
-          logging.info(`Sent ${totalSent} ${name} events to Tinybird`);
-        }
-      } catch (err) {
-        logging.error(err);
-      }
-    };
-
-    setTimeout(async () => {
-      await syncAutomationEvents({
-        table: 'automation_runs',
-        datasource: 'automation_run_events',
-        name: 'automation run',
-      });
-      await syncAutomationEvents({
-        table: 'automation_run_steps',
-        datasource: 'automation_run_step_events',
-        name: 'automation run step',
-      });
-    }, 10_000);
-  }
-
   // Resume any newsletter sends interrupted by a prior container shutdown.
   // Runs before activitypub.init so an activitypub failure can't disable recovery.
   try {
@@ -608,6 +521,16 @@ async function initBackgroundServices({ config }) {
     logging.error(err);
   }
 
+  if (config.get('backgroundJobs:tinybirdSync')) {
+    try {
+      const tinybirdSync = require('./server/services/tinybird-sync');
+      await tinybirdSync.scheduleTinybirdSyncJob(jobsService);
+    } catch (err) {
+      const logging = require('@tryghost/logging');
+      logging.error(err);
+    }
+  }
+
   const activitypub = require('./server/services/activitypub');
   await activitypub.init();
   // Load email analytics recurring jobs

ghost/core/core/server/data/exporter/table-lists.js

@@ -76,6 +76,7 @@ const BACKUP_TABLES = [
   'automation_runs',
   'welcome_email_automation_runs',
   'welcome_email_automated_emails',
+  'tinybird_syncs',
 ];
 
 // NOTE: exposing only tables which are going to be included in a "default" export file

ghost/core/core/server/data/migrations/versions/6.63/2026-09-02-18-23-36-add-tinybird-syncs-table.js

@@ -0,0 +1,9 @@
+const { addTable } = require('../../utils');
+
+module.exports = addTable('tinybird_syncs', {
+  id: { type: 'string', maxlength: 24, nullable: false, primary: true },
+  table_name: { type: 'string', maxlength: 191, nullable: false, unique: true },
+  last_synced_updated_at: { type: 'dateTime', nullable: false },
+  created_at: { type: 'dateTime', nullable: false },
+  updated_at: { type: 'dateTime', nullable: true },
+});

ghost/core/core/server/data/migrations/versions/6.63/2026-09-02-18-24-50-add-automation-updated-at-indexes.js

@@ -0,0 +1,6 @@
+const { combineTransactionalMigrations, createAddIndexMigration } = require('../../utils');
+
+module.exports = combineTransactionalMigrations(
+  createAddIndexMigration('automation_runs', ['updated_at']),
+  createAddIndexMigration('automation_run_steps', ['updated_at']),
+);

ghost/core/core/server/data/schema/schema.js

@@ -2258,7 +2258,7 @@ module.exports = {
       nullable: false,
       validations: { isEmail: true },
     },
-    '@@INDEXES@@': [['automation_id', 'created_at']],
+    '@@INDEXES@@': [['automation_id', 'created_at'], ['updated_at']],
   },
   automation_run_steps: {
     id: { type: 'string', maxlength: 24, nullable: false, primary: true },
@@ -2302,7 +2302,14 @@ module.exports = {
     },
     locked_by: { type: 'string', maxlength: 191, nullable: true },
     locked_at: { type: 'dateTime', nullable: true },
-    '@@INDEXES@@': [['status', 'ready_at', 'created_at', 'id']],
+    '@@INDEXES@@': [['status', 'ready_at', 'created_at', 'id'], ['updated_at']],
+  },
+  tinybird_syncs: {
+    id: { type: 'string', maxlength: 24, nullable: false, primary: true },
+    table_name: { type: 'string', maxlength: 191, nullable: false, unique: true },
+    last_synced_updated_at: { type: 'dateTime', nullable: false },
+    created_at: { type: 'dateTime', nullable: false },
+    updated_at: { type: 'dateTime', nullable: true },
   },
   welcome_email_automated_emails: {
     id: { type: 'string', maxlength: 24, nullable: false, primary: true },

ghost/core/core/server/data/tinybird/README.md

@@ -108,6 +108,18 @@ Keep in mind that as you update fixtures, it will rebuild data, but materialized
 data will not be cleared from them. One way to approach this to make sure data is consistent is to truncate all data
 sources before adding test data to it.
 
+### Automation data sync
+
+Ghost copies `automation_runs` and `automation_run_steps` into the `automation_run_events` and
+`automation_run_step_events` data sources with a recurring job (`core/server/services/tinybird-sync`).
+It runs every five minutes when Tinybird is configured, sending only rows updated since the
+watermark stored in the `tinybird_syncs` table. Set `backgroundJobs.tinybirdSync` to `false` in config
+to turn the job off.
+
+To force a full backfill, for example after pointing Ghost at a different Tinybird workspace or
+truncating the data sources, delete the matching rows from `tinybird_syncs`; the next run starts from
+the beginning.
+
 ### Architecture
 
 [See full documentation regarding analytics architecture in following document](ARCHITECTURE.md)

ghost/core/core/server/services/automations/automations-api.ts

@@ -11,9 +11,7 @@ const { knex } = require('../../data/db');
 const domainEvents = require('@tryghost/domain-events');
 const labs = require('../../../shared/labs');
 const config = require('../../../shared/config');
-const settingsCache = require('../../../shared/settings-cache');
 const lexicalLib = require('../../lib/lexical');
-const TinybirdServiceWrapper = require('../tinybird');
 
 const MAX_AUTOMATION_ACTIONS = 20;
 
@@ -75,58 +73,7 @@ const repository = createDatabaseAutomationsRepository({
 });
 
 export async function browse() {
-  console.time('@@@@ full browse')
-  const result = await repository.browse();
-  const tinybirdConfig = config.get('tinybird:stats');
-  const endpoint = tinybirdConfig.local?.enabled
-    ? tinybirdConfig.local.endpoint
-    : tinybirdConfig.endpoint;
-  const siteUuid = tinybirdConfig.id || settingsCache.get('site_uuid');
-
-  TinybirdServiceWrapper.init();
-  const token = TinybirdServiceWrapper.instance.getToken().token;
-
-  console.time('@@@@ tinybird req')
-  const response = await fetch(
-    `${endpoint}/v0/pipes/api_automation_browse_stats.json?site_uuid=${siteUuid}`,
-    {
-      headers: {
-        Authorization: `Bearer ${token}`,
-      },
-    },
-  );
-  console.timeEnd('@@@@ tinybird req')
-
-  if (!response.ok) {
-    throw new errors.InternalServerError({
-      message: `Tinybird API error: ${response.status} - ${await response.text()}`,
-    });
-  }
-
-  const { data } = (await response.json()) as {
-    data: Array<{
-      automation_id: string;
-      last_run_created_at: string;
-      total_run_count: number | string;
-      in_progress_run_count: number | string;
-    }>;
-  };
-  const statsByAutomationId = new Map(data.map((stats) => [stats.automation_id, stats]));
-
-  result.data = result.data.map((automation) => {
-    const stats = statsByAutomationId.get(automation.id);
-    return {
-      ...automation,
-      stats: {
-        last_run_created_at: stats ? new Date(stats.last_run_created_at) : null,
-        total_run_count: Number(stats?.total_run_count ?? 0),
-        in_progress_run_count: Number(stats?.in_progress_run_count ?? 0),
-      },
-    };
-  });
-
-  console.timeEnd('@@@@ full browse')
-  return result;
+  return await repository.browse();
 }
 
 export async function read(automationId: string) {

ghost/core/core/server/services/automations/database-automations-repository.ts

@@ -62,6 +62,12 @@ interface AutomationRow {
   updated_at: DatabaseDate;
 }
 
+interface AutomationBrowseRow extends AutomationRow {
+  last_run_created_at: DatabaseDate | null;
+  total_run_count: string | number | null;
+  in_progress_run_count: string | number | null;
+}
+
 interface ActionRow {
   id: string;
   type: 'wait' | 'send_email';
@@ -1064,9 +1070,32 @@ async function loadAutomationBySlug(
   return row ?? null;
 }
 
-async function loadAutomations(trx: Knex.Transaction): Promise<AutomationRow[]> {
+async function loadAutomations(trx: Knex.Transaction): Promise<AutomationBrowseRow[]> {
+  const inProgressRuns = trx('automation_run_steps')
+    .distinct('automation_run_id')
+    .where('status', 'pending')
+    .as('in_progress_runs');
+  const runStats = trx('automation_runs')
+    .select('automation_runs.automation_id')
+    .max({ last_run_created_at: 'automation_runs.created_at' })
+    .count({ total_run_count: '*' })
+    .count({ in_progress_run_count: 'in_progress_runs.automation_run_id' })
+    .leftJoin(inProgressRuns, 'automation_runs.id', 'in_progress_runs.automation_run_id')
+    .groupBy('automation_runs.automation_id')
+    .as('run_stats');
   return await trx('automations')
-    .select('id', 'slug', 'name', 'status', 'created_at', 'updated_at')
+    .select(
+      'automations.id',
+      'automations.slug',
+      'automations.name',
+      'automations.status',
+      'automations.created_at',
+      'automations.updated_at',
+      'run_stats.last_run_created_at',
+      'run_stats.total_run_count',
+      'run_stats.in_progress_run_count',
+    )
+    .leftJoin(runStats, 'automations.id', 'run_stats.automation_id')
     .orderBy('automations.name');
 }
 
@@ -1465,8 +1494,17 @@ function buildAutomationSummary(automation: AutomationRow): AutomationSummary {
   };
 }
 
-function buildAutomationBrowseResult(automation: AutomationRow): AutomationBrowseResult {
-  return buildAutomationSummary(automation);
+function buildAutomationBrowseResult(automation: AutomationBrowseRow): AutomationBrowseResult {
+  return {
+    ...buildAutomationSummary(automation),
+    stats: {
+      last_run_created_at: automation.last_run_created_at
+        ? fromDatabaseDate(automation.last_run_created_at)
+        : null,
+      total_run_count: Number(automation.total_run_count ?? 0),
+      in_progress_run_count: Number(automation.in_progress_run_count ?? 0),
+    },
+  };
 }
 
 function serializeDate(date: DatabaseDate) {

ghost/core/core/server/services/automations/poll.ts

@@ -152,16 +152,16 @@ const processStep = async ({
   // NOTE: This will change once we support additional automation triggers.
   const memberStatus = slugToMemberStatus.get(step.automation_slug);
   if (!memberStatus) {
-    // logging.error(
-    //   {
-    //     system: {
-    //       event: 'automations.poll.unknown_slug',
-    //       slug: step.automation_slug,
-    //       step_id: step.id,
-    //     },
-    //   },
-    //   `[AUTOMATIONS] Unknown automation slug: ${step.automation_slug}`,
-    // );
+    logging.error(
+      {
+        system: {
+          event: 'automations.poll.unknown_slug',
+          slug: step.automation_slug,
+          step_id: step.id,
+        },
+      },
+      `[AUTOMATIONS] Unknown automation slug: ${step.automation_slug}`,
+    );
     await automationsApi.markStepTerminal(step, 'failed');
     return null;
   }

ghost/core/core/server/services/jobs-service/register-job-handlers.ts

@@ -10,6 +10,8 @@ import * as contentImport from '../content-import';
 import UpdateCheckJob from '../update-check/jobs/update-check-job';
 import type MentionController from '../mentions/mention-controller';
 import ProcessWebmentionJob from '../mentions/process-webmention-job';
+import TinybirdSyncJob from '../tinybird-sync/tinybird-sync-job';
+import * as tinybirdSync from '../tinybird-sync';
 
 const updateCheck = require('../update-check');
 
@@ -58,4 +60,8 @@ export default function registerJobHandlers({
   jobsService.handle(ProcessWebmentionJob, async (job) => {
     await mentionsController.processWebmention(job);
   });
+
+  jobsService.handle(TinybirdSyncJob, async () => {
+    await tinybirdSync.run();
+  });
 }