Task Authoring Basics

/// <reference types="../.compose/types.d.ts" />

/// <reference types="../../.compose/types.d.ts" /> 

export async function main(
  // injected by compose
  context: TaskContext, 
  // update this with whatever payload your task will receive if your task supports http triggers
  // this will be undefined when your task is triggered with cron
  payload?: Record<string, unknown> 
): Promise<unknown> {
  // Task implementation
}

export type LogEvent = {
  code: string;
  message: string;
  data: string;
  task?: string;
  runId?: string;
};

export type ContextFunctionRetryConfig = {
  max_attempts: number;
  initial_interval_ms: number;
  backoff_factor: number;
};

export type Chain = {
  id: number;
  name: string;
  testnet: boolean;
  nativeCurrency: {
    name: string;
    symbol: string;
    decimals: number;
  };
  rpcUrls: {
    public: { http: string[] };
    default: { http: string[] };
  };
  blockExplorers: {
    default: { name: string; url: string };
  };
  contracts?: Record<string, { address: string }>;
};

export type ScalarIndexType = "text" | "numeric" | "boolean" | "timestamptz";

export interface FindOptions {
  limit?: number;
  offset?: number;
}

export type Filter = Record<string, string | unknown>;

export type WithId<T> = T & { id: string };

export interface Collection<TDoc = unknown> {
  readonly name: string;
  insertOne(doc: TDoc, opts?: { id?: string }): Promise<{ id: string }>;
  findOne(filter: Filter): Promise<WithId<TDoc> | null>;
  findMany(filter: Filter, options?: FindOptions): Promise<Array<WithId<TDoc>>>;
  list(options?: FindOptions): Promise<Array<WithId<TDoc>>>;
  getById(id: string): Promise<WithId<TDoc> | null>;
  setById(id: string, doc: TDoc, opts?: { upsert?: boolean }): Promise<{ id: string; upserted?: boolean; matched?: number }>;
  deleteById(id: string): Promise<{ deletedCount: number }>;
  createScalarIndex(path: string, opts: { type: ScalarIndexType; unique?: boolean }): Promise<void>;
  drop(): Promise<void>;
}

export type Address = `0x${string}`;

export interface WalletConfig {
  name?: string; // defaults to "default"
  privateKey?: string;
  sponsorGas?: boolean; // defaults to true if no privateKey and false if privateKey
}

type ReplayOnReorg = {
  type: "replay";
};

type LogOnReorg = {
  type: "log";
  logLevel?: "error" | "info" | "warn"; // defaults to "error"
};

type Custom = {
  type: "custom",
  // your task will be sent with a payload the full transaction minus gas and nonce
  task: string; 
}

type OnReorgOptions = ReplayOnReorg | LogOnReorg | Custom;

export interface TransactionConfirmation {
  // this the number of block confirmations before we resolve the promise
  // i.e. "wait 5 blocks before proceeding to the next step in my task"
  confirmations?: number; 
  onReorg?: {
    action: OnReorgOptions;
    depth: number;
  };
}

export interface IWallet {
  readonly name: string;
  readonly address: string;
  writeContract(
    chain: Chain,
    contractAddress: Address,
    functionSig: string,
    args: unknown[],
    confirmation?: TransactionConfirmation,
    retryConfig?: ContextFunctionRetryConfig
  ): Promise<{ hash: string }>;
  readContract: (
    chain: Chain,
    contractAddress: Address,
    functionSig: string,
    args: unknown[],
    retryConfig?: ContextFunctionRetryConfig
  ) => Promise<unknown>;
  simulate: (
    chain: Chain,
    contractAddress: Address,
    functionSig: string,
    args: unknown[],
    retryConfig?: ContextFunctionRetryConfig
  ) => Promise<unknown>;
}

export type TaskContext = {
  env: Record<"local" | "cloud", Record<string, string>>;
  callTask: <T = unknown>(
    taskName: string,
    args: Record<string, unknown>,
    retryConfig?: ContextFunctionRetryConfig,
  ) => Promise<T>;
  fetch: <T = unknown>(
    url: string,
    bodyOrRetryConfig?: Record<string, unknown> | ContextFunctionRetryConfig,
    retryConfig?: ContextFunctionRetryConfig
  ) => Promise<T | undefined>;
  logEvent: (
    event: LogEvent,
    retryConfig?: ContextFunctionRetryConfig
  ) => Promise<void>;
  evm: {
    chains: Record<string, Chain>;
    wallet: (config: WalletConfig) => Promise<IWallet>;
  }
  collection: <T>(name: string) => Promise<Collection<T>>;
};

tasks:
  - name: "hourly_sync"
    path: "./tasks/sync.ts"
    cron:
      expression: "0 * * * *" # Every hour at minute 0

/// <reference types="../../.compose/types.d.ts" /> 

export async function main({ fetch, stage, logEvent }) {
  await logEvent({
    code: "SYNC_STARTED",
    message: "Hourly sync initiated",
    data: JSON.stringify({ timestamp: Date.now() }),
  });

  // Perform sync operations
  const data = await fetch("https://api.example.com/sync");
  await stage.set(
    "last_sync",
    JSON.stringify({
      timestamp: Date.now(),
      recordCount: data.length,
    })
  );

  return { success: true, synced: data.length };
}

tasks:
  - name: "unreliable_task"
    path: "./tasks/flaky.ts"
    retry_config:
      max_attempts: 5
      initial_interval_ms: 2000
      backoff_factor: 1.5

  max_attempts: 5
  initial_interval_ms: 1000
  backoff_factor: 2

/// <reference types="../../.compose/types.d.ts" /> 

type Dog = {
  breed: string;
  color: string;
}

export async function main({ stage, fetch, logEvent }) {
  const dogsCollection = await stage.collection<Dog>("dogs");

  // Get existing state
  const allDogs = await dogsCollection.list({ offset: 0, limit: 20 });
  const brownDogs = await dogsCollection.findMany({ color: "brown" });

  // Get some data
  const newData = await fetch<Dog>("https://api.dogs.com/v1/dogs/labrador");

  // Update state
  await dogsCollection.insertOne(newData);

  return {
    success: true,
  };
}