Extensibility, Resilience & Scalability

// NEW task - zero changes to the code that initiates payments
import { task, tasks } from '@trigger.dev/sdk/v3';

export const fraudDetection = task({
  id: 'fraud-detection',
  queue: { concurrencyLimit: 10 },
  run: async (payload: { paymentId: string; amount: number }) => {
    const riskScore = await analyzeFraudRisk(payload);

    if (riskScore > 0.8) {
      // Fan out to a follow-up task instead of a broker "sendEvent"
      await tasks.trigger('alert-fraud-team', { ...payload, riskScore });
    }

    return { paymentId: payload.paymentId, riskScore };
  },
});

// New IoT feature - add a NEW task; nothing existing changes
import { task } from '@trigger.dev/sdk/v3';

export const iotDataProcessor = task({
  id: 'iot-data-processor',
  run: async (payload: {
    sensorId: string;
    reading: number;
    timestamp: string;
    location: string;
  }) => {
    return await storeIoTReading(payload);
  },
});

// The producer adds one line to fan the new work out:
//   await iotDataProcessor.trigger({ sensorId, reading, timestamp, location });
// Existing tasks are never invoked with this payload - no impact.

// Version 1: Original schema
const WorkspaceCreatedV1 = z.object({
  workspaceId: z.string(),
  ownerId: z.string(),
  createdAt: z.date()
});

// Version 2: Add optional fields (backward compatible)
const WorkspaceCreatedV2 = z.object({
  workspaceId: z.string(),
  ownerId: z.string(),
  createdAt: z.date(),
  correlationId: z.string().optional(),  // NEW: optional
  source: z.string().optional()           // NEW: optional
});

// Old consumers still work (ignore new fields)
// New consumers can use new fields

// Conceptual: a feature-owned task that writes to its own geospatial tables.
import { task } from '@trigger.dev/sdk/v3';

export const updateUserLocation = task({
  id: 'update-user-location',
  run: async (payload: { userId: string; lat: number; lng: number }) => {
    // Writes to geolocation-owned tables (e.g. a PostGIS extension),
    // without touching unrelated schemas.
    return await storeUserLocation(payload);
  },
});

// In Tuturuuu today this lives in the shared Supabase project; isolation is
// schema/RLS-based rather than a physically separate database.

// apps/web/src/components/dashboard/NewWidget.tsx
'use client';

import { useMutation } from '@tanstack/react-query';

export function NewDashboardWidget() {
  // Use TanStack Query for client mutations - never fetch in useEffect.
  const refresh = useMutation({
    mutationFn: () =>
      fetch('/api/v1/widgets/refresh', {
        method: 'POST',
        body: JSON.stringify({ widgetId: 'new-widget' }),
      }),
  });

  return (
    <div className="widget">
      <button onClick={() => refresh.mutate()}>Refresh Data</button>
      {/* Widget UI */}
    </div>
  );
}

// Backend route: apps/web/src/app/api/v1/widgets/refresh/route.ts
import { tasks } from '@trigger.dev/sdk/v3';

export async function POST(request: Request) {
  const { widgetId } = await request.json();

  // Kick off background work by triggering a task (no broker publish).
  await tasks.trigger('refresh-widget', { widgetId });

  return Response.json({ success: true });
}

import { task, tasks } from '@trigger.dev/sdk/v3';

// The caller continues working even if processing is backed up.
await tasks.trigger('process-report', { reportId, data });
// Returns immediately - it does not wait for the run to complete.

// The task drains its backlog at its own sustainable pace.
export const processReport = task({
  id: 'process-report',
  queue: { concurrencyLimit: 5 },
  run: async (payload: { reportId: string; data: unknown }) => {
    return await generateReport(payload);
  },
});

// SYNCHRONOUS (problematic):
async function createWorkspace(data) {
  const workspace = await db.createWorkspace(data);

  // Blocks if email service is slow (5 seconds)
  await sendWelcomeEmail(workspace.ownerId);  // BLOCKS HERE

  // Blocks if provisioning is slow (10 seconds)
  await provisionResources(workspace.id);      // BLOCKS HERE

  return workspace;  // User waits 15+ seconds
}

// EVENT-DRIVEN (resilient):
import { tasks } from '@trigger.dev/sdk/v3';

async function createWorkspace(data) {
  const workspace = await db.createWorkspace(data);

  // Non-blocking - enqueues background work and returns immediately
  await tasks.trigger('workspace-created-followups', { workspaceId: workspace.id });

  return workspace;  // User gets a response in <1 second
}

// Tasks run asynchronously.
// If the email step is slow, it does not affect the user-facing response.

import { task } from '@trigger.dev/sdk/v3';

export const sendWelcomeEmail = task({
  id: 'send-welcome-email',
  run: async (payload: { userId: string; email: string }) => {
    const { userId, email } = payload;

    // Idempotent: check if already sent before sending.
    const existingEmail = await db.sentEmails.findOne({
      userId,
      type: 'welcome',
    });

    if (existingEmail) {
      // Already sent - skip (idempotent behavior).
      return { skipped: true };
    }

    await sendEmail({ to: email, template: 'welcome' });

    await db.sentEmails.insert({ userId, type: 'welcome', sentAt: new Date() });

    return { sent: true };
  },
});

// If this run is retried, the user does not get a duplicate email.
// Triggering with an idempotencyKey also deduplicates the run itself:
//   sendWelcomeEmail.trigger(payload, { idempotencyKey: `welcome-${userId}` });

import { task, tasks } from '@trigger.dev/sdk/v3';

export const processPayment = task({
  id: 'process-payment',
  // Real v4 retry policy: bounded, exponential backoff.
  retry: { maxAttempts: 3, factor: 2, minTimeoutInMs: 1000 },
  run: async (payload: { id: string }) => {
    return await chargePaymentMethod(payload);
  },
  // Runs after all retries are exhausted - the "dead-letter" hand-off.
  handleError: async (payload, error) => {
    await tasks.trigger('payment-dead-letter', {
      originalPayload: payload,
      error: error instanceof Error ? error.message : 'Unknown error',
    });
  },
});

// Dedicated task that records failures and alerts for manual review.
export const paymentDeadLetter = task({
  id: 'payment-dead-letter',
  run: async (payload: { originalPayload: { id: string }; error: string }) => {
    await recordFailedPayment(payload);
    await alertOnCallTeam(payload.originalPayload.id);
  },
});

DISASTER: a derived read model (e.g. a cache or projection) is corrupted by a bug.
1. Stop the consumer that builds the read model.
2. Deploy the fixed projection logic.
3. Discard the corrupted derived state.
4. Re-process the durable source events in order to rebuild the read model.

// Conceptual rebuild loop - a plain task that re-applies events you have
// stored durably. There is no built-in Trigger.dev "replay by date" trigger;
// you supply the events yourself (e.g. from an append-only table).
import { task } from '@trigger.dev/sdk/v3';

export const rebuildUserProjection = task({
  id: 'rebuild-user-projection',
  run: async (payload: { event: UserEvent }) => {
    const { event } = payload;
    if (event.type === 'user.registered') await db.users.insert(event);
    else if (event.type === 'user.updated') await db.users.update(event.userId, event);
    else if (event.type === 'user.deleted') await db.users.delete(event.userId);
  },
});

import { task } from '@trigger.dev/sdk/v3';

export const processAnalytics = task({
  id: 'process-analytics',
  // Higher concurrency = more parallel runs = more throughput.
  queue: { concurrencyLimit: 50 },
  run: async (payload: { userId: string; activity: unknown }) => {
    return await aggregateMetrics(payload);
  },
});

// Trigger.dev schedules many runs of this task in parallel up to the limit,
// so raising the concurrency limit raises throughput without code changes.

import { task } from '@trigger.dev/sdk/v3';

export const updateWorkspace = task({
  id: 'update-workspace',
  // Serialize runs per workspace (one in flight per concurrencyKey),
  // while different workspaces run in parallel.
  queue: { concurrencyLimit: 25 },
  run: async (payload: { workspaceId: string; data: unknown }) => {
    return await applyWorkspaceUpdate(payload);
  },
});

// Trigger per workspace so same-workspace runs are ordered:
//   updateWorkspace.trigger(payload, { concurrencyKey: payload.workspaceId });
//
// Workspace A runs: serialized (ordered)
// Workspace B runs: serialized (ordered), in parallel with A
// Workspace C runs: serialized (ordered), in parallel with A & B

import { task } from '@trigger.dev/sdk/v3';

// Heavy, high-volume work: allow many parallel runs.
export const processAnalytics = task({
  id: 'process-analytics',
  queue: { concurrencyLimit: 100 },
  run: async (payload) => aggregateMetrics(payload),
});

// Low-volume, rate-sensitive work: keep it narrow.
export const sendReport = task({
  id: 'send-report',
  queue: { concurrencyLimit: 2 },
  run: async (payload) => deliverReport(payload),
});

// Each task scales independently via its own concurrency limit -
// no need to scale the whole app to speed up one workload.

import { task, tasks } from '@trigger.dev/sdk/v3';

// COMMAND: write the authoritative record, then trigger the projection.
async function updateWorkspace(id: string, data: unknown) {
  await db.workspaces.update(id, data);
  await tasks.trigger('update-workspace-cache', { workspaceId: id, data });
}

// QUERY: maintain a read-optimized cache.
export const updateWorkspaceCache = task({
  id: 'update-workspace-cache',
  run: async (payload: { workspaceId: string; data: Record<string, unknown> }) => {
    await redis.set(
      `workspace:${payload.workspaceId}`,
      JSON.stringify({
        ...payload.data,
        memberCount: await getMemberCount(payload.workspaceId),
        lastActivity: new Date(),
      })
    );
  },
});

// Reads hit the cache (fast); writes go to Postgres (consistent).