Set up Background sync using Trigger.dev

.gitignore

@@ -25,9 +25,9 @@ dist
 # debug
 npm-debug.log*
 yarn-debug.log*
-yarn-error.log*
 
 # local env files
+.env
 .env.local
 .env.development.local
 .env.test.local
@@ -38,4 +38,10 @@ yarn-error.log*
 
 # Typescript
 *.tsbuildinfo
-next-env.d.ts
+next-env.d.ts
+.trigger.env
+.env.local
+.env.*.local
+
+# Trigger.dev
+.trigger/

apps/db/supabase/migrations/20250101000000_add_calendar_sync_coordination.sql

@@ -0,0 +1,53 @@
+-- Create table to coordinate calendar sync operations
+create table "public"."workspace_calendar_sync_coordination" (
+    "ws_id" uuid not null,
+    "last_upsert" timestamp with time zone not null default now(),
+    "created_at" timestamp with time zone default now(),
+    "updated_at" timestamp with time zone default now()
+);
+
+-- Add primary key
+alter table "public"."workspace_calendar_sync_coordination" 
+add constraint "workspace_calendar_sync_coordination_pkey" PRIMARY KEY (ws_id);
+
+-- Add foreign key to workspaces
+alter table "public"."workspace_calendar_sync_coordination" 
+add constraint "workspace_calendar_sync_coordination_ws_id_fkey" 
+FOREIGN KEY (ws_id) REFERENCES workspaces(id) ON DELETE CASCADE;
+
+-- Enable RLS
+alter table "public"."workspace_calendar_sync_coordination" enable row level security;
+
+-- Create policy for workspace members
+create policy "Enable access for workspace members" on "public"."workspace_calendar_sync_coordination" 
+as permissive for all to authenticated using (
+    EXISTS (
+        SELECT 1 FROM workspace_members 
+        WHERE workspace_members.ws_id = workspace_calendar_sync_coordination.ws_id
+        AND workspace_members.user_id = auth.uid()
+    )
+) with check (
+    EXISTS (
+        SELECT 1 FROM workspace_members 
+        WHERE workspace_members.ws_id = workspace_calendar_sync_coordination.ws_id
+        AND workspace_members.user_id = auth.uid()
+    )
+);
+
+-- Create function to update the updated_at timestamp
+CREATE OR REPLACE FUNCTION update_workspace_calendar_sync_coordination_updated_at()
+RETURNS TRIGGER AS $$
+BEGIN
+    NEW.updated_at = now();
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Create trigger to automatically update updated_at
+CREATE TRIGGER update_workspace_calendar_sync_coordination_updated_at
+    BEFORE UPDATE ON workspace_calendar_sync_coordination
+    FOR EACH ROW
+    EXECUTE FUNCTION update_workspace_calendar_sync_coordination_updated_at();
+
+-- Enable audit tracking
+select audit.enable_tracking('public.workspace_calendar_sync_coordination'::regclass);

apps/web/src/app/[locale]/(dashboard)/[wsId]/calendar/active-sync.tsx

@@ -54,6 +54,12 @@ const InnerComponent = () => {
     }
   };
 
+  const triggerHelloWorld = async () => {
+    const res = await fetch('/api/hello-world');
+    const data = await res.json();
+    console.log(data);
+  };
+
   return (
     <div>
       <div className="mb-4 flex gap-2">
@@ -85,6 +91,7 @@ const InnerComponent = () => {
         >
           Switch to month
         </Button>
+        <Button onClick={() => triggerHelloWorld()}>Trigger Hello World</Button>
       </div>
 
       {/* Add sync progress bar when syncing */}

apps/web/src/app/api/hello-world/route.ts

@@ -0,0 +1,16 @@
+// Next.js API route support: https://nextjs.org/docs/api-routes/introduction
+import { tasks } from '@trigger.dev/sdk/v3';
+import type { helloWorldTask } from '@tuturuuu/trigger/example';
+import { NextResponse } from 'next/server';
+
+//tasks.trigger also works with the edge runtime
+//export const runtime = "edge";
+
+export async function GET() {
+  const handle = await tasks.trigger<typeof helloWorldTask>(
+    'hello-world',
+    'James'
+  );
+
+  return NextResponse.json(handle);
+}

package.json

@@ -64,6 +64,7 @@
     "@tuturuuu/eslint-config": "workspace:*",
     "@tuturuuu/typescript-config": "workspace:*",
     "@vitest/coverage-v8": "^3.2.3",
+    "concurrently": "^9.1.2",
     "eslint": "^9.28.0",
     "prettier": "^3.5.3",
     "prettier-eslint": "^16.4.2",

packages/trigger/.env.example

@@ -0,0 +1,43 @@
+# Required environment variables
+# for both development and production
+NEXT_PUBLIC_SUPABASE_URL=YOUR_SUPABASE_URL
+NEXT_PUBLIC_SUPABASE_ANON_KEY=YOUR_SUPABASE_ANON_KEY
+SUPABASE_SERVICE_KEY=YOUR_SUPABASE_SERVICE_KEY
+
+# Optional API keys
+OPENAI_API_KEY=YOUR_OPENAI_API_KEY 
+ANTHROPIC_API_KEY=YOUR_ANTHROPIC_API_KEY
+GOOGLE_GENERATIVE_AI_API_KEY=YOUR_GOOGLE_GENERATIVE_AI_API_KEY
+
+# Google Vertex AI credentials
+GOOGLE_VERTEX_PROJECT=YOUR_GOOGLE_VERTEX_PROJECT
+GOOGLE_VERTEX_LOCATION=YOUR_GOOGLE_VERTEX_LOCATION
+GOOGLE_APPLICATION_CREDENTIALS=YOUR_GOOGLE_APPLICATION_CREDENTIALS
+
+# Google Calendar API credentials
+GOOGLE_CLIENT_ID=YOUR_GOOGLE_CLIENT_ID
+GOOGLE_CLIENT_SECRET=YOUR_GOOGLE_CLIENT_SECRET
+GOOGLE_REDIRECT_URI=YOUR_GOOGLE_REDIRECT_URI
+
+# AWS Credentials
+AWS_REGION=YOUR_AWS_REGION
+AWS_ACCESS_KEY_ID=YOUR_AWS_ACCESS_KEY_ID
+AWS_SECRET_ACCESS_KEY=YOUR_AWS_SECRET
+SOURCE_NAME=YOUR_SOURCE_NAME
+SOURCE_EMAIL=YOUR_SOURCE_EMAIL
+
+DEEPGRAM_ENV=YOUR_DEEPGRAM_ENV
+DEEPGRAM_API_KEY=YOUR_DEEPGRAM_API_KEY
+
+MODAL_TOKEN_ID=YOUR_MODAL_TOKEN_ID
+MODAL_TOKEN_SECRET=YOUR_MODAL_TOKEN_SECRET
+
+CF_ACCOUNT_ID=YOUR_CF_ACCOUNT_ID
+CF_API_TOKEN=YOUR_CF_API_TOKEN
+
+# Infrastructure Credentials
+SCRAPER_URL=YOUR_SCRAPER_URL
+AURORA_EXTERNAL_URL=YOUR_AURORA_EXTERNAL_URL
+AURORA_EXTERNAL_WSID=YOUR_AURORA_EXTERNAL_WSID
+PROXY_API_KEY=YOUR_PROXY_API_KEY
+NEXT_PUBLIC_PROXY_API_KEY=YOUR_NEXT_PUBLIC_PROXY_API_KEY

packages/trigger/example.ts

@@ -0,0 +1,16 @@
+import { logger, task, wait } from '@trigger.dev/sdk/v3';
+
+export const helloWorldTask = task({
+  id: 'hello-world',
+  // Set an optional maxDuration to prevent tasks from running indefinitely
+  maxDuration: 300, // Stop executing after 300 secs (5 mins) of compute
+  run: async (payload: any, { ctx }) => {
+    logger.log('Hello, world!', { payload, ctx });
+
+    await wait.for({ seconds: 5 });
+
+    return {
+      message: 'Hello, world!',
+    };
+  },
+});

packages/trigger/first-scheduled-task.ts

@@ -0,0 +1,35 @@
+import { schedules } from '@trigger.dev/sdk/v3';
+
+export const firstScheduledTask = schedules.task({
+  id: 'first-scheduled-task',
+  cron: {
+    // every 1 minute
+    pattern: '*/1 * * * *',
+  },
+  run: async (payload) => {
+    //when the task was scheduled to run
+    //note this will be slightly different from new Date() because it takes a few ms to run the task
+    console.log(payload.timestamp); //is a Date object
+
+    //when the task was last run
+    //this can be undefined if it's never been run
+    console.log(payload.lastTimestamp); //is a Date object or undefined
+
+    //the timezone the schedule was registered with, defaults to "UTC"
+    //this is in IANA format, e.g. "America/New_York"
+    //See the full list here: https://cloud.trigger.dev/timezones
+    console.log(payload.timezone); //is a string
+
+    //the schedule id (you can have many schedules for the same task)
+    //using this you can remove the schedule, update it, etc
+    console.log(payload.scheduleId); //is a string
+
+    //you can optionally provide an external id when creating the schedule
+    //usually you would set this to a userId or some other unique identifier
+    //this can be undefined if you didn't provide one
+    console.log(payload.externalId); //is a string or undefined
+
+    //the next 5 dates this task is scheduled to run
+    console.log(payload.upcoming); //is an array of Date objects
+  },
+});

packages/trigger/google-calendar-background-sync.ts

@@ -0,0 +1,161 @@
+import { createClient } from '@supabase/supabase-js';
+import { schedules } from '@trigger.dev/sdk/v3';
+import {
+  BACKGROUND_SYNC_RANGE,
+  updateLastUpsert,
+} from '@tuturuuu/utils/calendar-sync-coordination';
+import dayjs from 'dayjs';
+import 'dotenv/config';
+import { OAuth2Client } from 'google-auth-library';
+import { google } from 'googleapis';
+
+export const googleCalendarBackgroundSync = schedules.task({
+  id: 'google-calendar-background-sync',
+  cron: {
+    // every 2 minutes
+    pattern: '*/2 * * * *',
+  },
+  run: async () => {
+    console.log(
+      'process.env.NEXT_PUBLIC_SUPABASE_URL',
+      process.env.NEXT_PUBLIC_SUPABASE_URL
+    );
+    // Initialize Supabase client inside the task function
+    const supabase = createClient(
+      process.env.NEXT_PUBLIC_SUPABASE_URL!,
+      process.env.SUPABASE_SERVICE_KEY!
+    );
+
+    await syncGoogleCalendarEvents(supabase);
+    console.log('Synced events from all linked Google accounts');
+  },
+});
+
+const getGoogleAuthClient = (tokens: {
+  access_token: string;
+  refresh_token?: string;
+}) => {
+  const oauth2Client = new OAuth2Client({
+    clientId: process.env.GOOGLE_CLIENT_ID,
+    clientSecret: process.env.GOOGLE_CLIENT_SECRET,
+    redirectUri: process.env.GOOGLE_REDIRECT_URI,
+  });
+
+  oauth2Client.setCredentials(tokens);
+  return oauth2Client;
+};
+
+const getColorFromGoogleColorId = (colorId?: string): string => {
+  const colorMap: Record<string, string> = {
+    '1': 'RED',
+    '2': 'GREEN',
+    '3': 'GRAY',
+    '4': 'PINK',
+    '5': 'YELLOW',
+    '6': 'ORANGE',
+    '8': 'CYAN',
+    '9': 'PURPLE',
+    '10': 'INDIGO',
+    '11': 'BLUE',
+  };
+  return colorId && colorMap[colorId] ? colorMap[colorId] : 'BLUE';
+};
+
+const syncGoogleCalendarEvents = async (supabase: any) => {
+  try {
+    // Fetch all wsId with auth tokens not null
+    const result = await supabase
+      .from('calendar_auth_tokens')
+      .select('ws_id, access_token, refresh_token');
+
+    const data = result.data;
+    const error = result.error;
+
+    if (error) {
+      console.error('Error fetching auth tokens:', error);
+      return [];
+    }
+
+    const googleTokens = data.map((item: any) => ({
+      ws_id: item.ws_id,
+      access_token: item.access_token,
+      refresh_token: item.refresh_token,
+    }));
+    // Type assertion for the tokens
+    const tokens = googleTokens as
+      | {
+          ws_id: string;
+          access_token: string;
+          refresh_token: string;
+        }[]
+      | null;
+
+    for (const token of tokens || []) {
+      const { ws_id, access_token, refresh_token } = token;
+      if (!access_token) {
+        console.error('No Google access token found for wsIds:', {
+          ws_id,
+          hasAccessToken: !!access_token,
+          hasRefreshToken: !!refresh_token,
+        });
+      }
+
+      try {
+        const auth = getGoogleAuthClient(token);
+        const calendar = google.calendar({ version: 'v3', auth });
+
+        const startOfCurrentWeek = dayjs().startOf('week');
+        const timeMin = startOfCurrentWeek.toDate();
+        const timeMax = startOfCurrentWeek
+          .add(BACKGROUND_SYNC_RANGE, 'day')
+          .toDate();
+
+        const response = await calendar.events.list({
+          calendarId: 'primary',
+          timeMin: timeMin.toISOString(), // from now
+          timeMax: timeMax.toISOString(), // to the next 4 weeks
+          singleEvents: true, // separate recurring events
+          orderBy: 'startTime',
+          maxResults: 1000,
+        });
+
+        const events = response.data.items || [];
+
+        // format the events to match the expected structure
+        const formattedEvents = events.map((event) => ({
+          google_event_id: event.id,
+          title: event.summary || 'Untitled Event',
+          description: event.description || '',
+          start_at: event.start?.dateTime || event.start?.date || '',
+          end_at: event.end?.dateTime || event.end?.date || '',
+          location: event.location || '',
+          color: getColorFromGoogleColorId(event.colorId ?? undefined),
+          ws_id: ws_id,
+          locked: false,
+        }));
+        console.log('ws_id', ws_id);
+        console.log('access_token', access_token);
+        console.log('refresh_token', refresh_token);
+        console.log('formattedEvents', formattedEvents);
+
+        // upsert the events in the database for this wsId
+        const { error } = await supabase
+          .from('workspace_calendar_events')
+          .upsert(formattedEvents, {
+            onConflict: 'google_event_id',
+          });
+        if (error) {
+          console.error('Error upserting events:', error);
+        } else {
+          // Update lastUpsert timestamp after successful upsert
+          await updateLastUpsert(ws_id, supabase);
+        }
+      } catch (error) {
+        console.error('Error fetching Google Calendar events:', error);
+      }
+    }
+  } catch (error) {
+    console.error('Error in fetchGoogleCalendarEvents:', error);
+    return [];
+  }
+};