Temporal Reference: Workflows, Activities, Signals, TypeScript SDK & Production Patterns

package app

import (
    "context"
    "time"
    "go.temporal.io/sdk/activity"
    "go.temporal.io/sdk/workflow"
)

// Activity: the side-effectful work (API calls, DB writes, etc.)
// Activities can fail and be retried — all I/O goes here.
func SendEmailActivity(ctx context.Context, to string, subject string) error {
    logger := activity.GetLogger(ctx)
    logger.Info("Sending email", "to", to)
    // Any I/O here — it will be retried on failure
    return emailClient.Send(to, subject, "...")
}

func ChargePaymentActivity(ctx context.Context, userID string, amount float64) (string, error) {
    // Returns a transaction ID
    return stripeClient.Charge(userID, amount)
}

// Workflow: the orchestration logic — MUST be deterministic
// No: time.Now(), rand, direct HTTP calls, goroutines (use workflow.Go), maps with ranging
func OrderWorkflow(ctx workflow.Context, order Order) error {
    logger := workflow.GetLogger(ctx)

    // Activity options: retry policy, timeouts:
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Second,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    100 * time.Second,
            MaximumAttempts:    5,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    // Execute activities — blocks until complete or error:
    var txID string
    if err := workflow.ExecuteActivity(ctx, ChargePaymentActivity, order.UserID, order.Total).Get(ctx, &txID); err != nil {
        return err
    }

    // workflow.Sleep is safe — survives restarts, doesn't block a thread:
    _ = workflow.Sleep(ctx, 5*time.Minute)

    if err := workflow.ExecuteActivity(ctx, SendEmailActivity, order.Email, "Order confirmed").Get(ctx, nil); err != nil {
        logger.Warn("Email failed, continuing", "error", err)
    }

    return nil
}

package main

import (
    "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"
)

func main() {
    // Connect to Temporal server (local or Temporal Cloud):
    c, err := client.Dial(client.Options{
        HostPort:  "localhost:7233",   // or "your-namespace.tmprl.cloud:7233"
        Namespace: "default",
    })
    defer c.Close()

    // Worker: polls for tasks on this task queue:
    w := worker.New(c, "order-task-queue", worker.Options{})
    w.RegisterWorkflow(OrderWorkflow)
    w.RegisterActivity(ChargePaymentActivity)
    w.RegisterActivity(SendEmailActivity)

    w.Run(worker.InterruptCh())
}

// Start a workflow (from your API server / anywhere):
we, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
    ID:        "order-" + orderID,     // idempotent key — re-running returns existing execution
    TaskQueue: "order-task-queue",
    WorkflowExecutionTimeout: 24 * time.Hour,
}, OrderWorkflow, order)

workflowID := we.GetID()
runID := we.GetRunID()

// Wait for result:
var result string
err = we.Get(ctx, &result)

// Query running workflow state:
resp, _ := c.QueryWorkflow(ctx, workflowID, runID, "getStatus")

// Signal a running workflow (external event):
c.SignalWorkflow(ctx, workflowID, runID, "paymentReceived", paymentPayload)

// Signals: receive external events in a running workflow
func SubscriptionWorkflow(ctx workflow.Context, userID string) error {
    // Register query handler (returns current state to callers):
    status := "active"
    workflow.SetQueryHandler(ctx, "getStatus", func() (string, error) {
        return status, nil
    })

    // Receive signal:
    cancelCh := workflow.GetSignalChannel(ctx, "cancel")
    renewCh  := workflow.GetSignalChannel(ctx, "renew")

    for {
        // workflow.Sleep survives server restarts:
        _ = workflow.Sleep(ctx, 30*24*time.Hour)   // 30 days

        // Check for signals (non-blocking selector):
        selector := workflow.NewSelector(ctx)
        selector.AddReceive(cancelCh, func(c workflow.ReceiveChannel, _ bool) {
            status = "cancelled"
        })
        selector.AddReceive(renewCh, func(c workflow.ReceiveChannel, _ bool) {
            status = "renewed"
        })

        if selector.HasPending() {
            selector.Select(ctx)
        }
        if status == "cancelled" {
            return nil
        }

        // Bill the user for another month:
        ao := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
            StartToCloseTimeout: 30 * time.Second,
        })
        workflow.ExecuteActivity(ao, BillUserActivity, userID).Get(ctx, nil)
    }
}

// Child workflows — split long workflows or fan out parallel work:
cwo := workflow.ChildWorkflowOptions{
    WorkflowID: "fulfillment-" + orderID,
    TaskQueue:  "fulfillment-queue",   // can run on a different worker pool
}
childCtx := workflow.WithChildOptions(ctx, cwo)
workflow.ExecuteChildWorkflow(childCtx, FulfillmentWorkflow, order).Get(ctx, nil)

// activities.ts — I/O goes here (NOT in workflow code):
import { Context } from "@temporalio/activity";

export async function sendEmail(to: string, subject: string): Promise {
  const { info } = Context.current();
  console.log(`Attempt ${info.attempt}: sending to ${to}`);
  await emailClient.send({ to, subject });
}

export async function chargePayment(userId: string, amount: number): Promise {
  return stripeClient.charge(userId, amount);
}

// workflow.ts — deterministic orchestration:
import { proxyActivities, sleep, defineSignal, setHandler } from "@temporalio/workflow";
import type * as activities from "./activities";

const { sendEmail, chargePayment } = proxyActivities({
  startToCloseTimeout: "10 seconds",
  retry: { maximumAttempts: 5, initialInterval: "1s", backoffCoefficient: 2 },
});

// Signals:
const cancelSignal = defineSignal("cancel");

export async function orderWorkflow(order: Order): Promise {
  let cancelled = false;
  setHandler(cancelSignal, () => { cancelled = true; });

  const txId = await chargePayment(order.userId, order.total);

  if (cancelled) return;

  await sleep("5 minutes");   // durable — survives restarts
  await sendEmail(order.email, `Order confirmed. Transaction: ${txId}`);
}

// worker.ts:
import { Worker } from "@temporalio/worker";
import * as activities from "./activities";
const worker = await Worker.create({
  taskQueue: "orders",
  workflowsPath: require.resolve("./workflow"),
  activities,
});
await worker.run();

# Local development with Temporal CLI:
# brew install temporal
temporal server start-dev --db-filename temporal.db
# Opens UI at http://localhost:8233

# Docker Compose for local dev:
# services:
#   temporal:
#     image: temporalio/auto-setup:1.24
#     ports: ["7233:7233", "8233:8233"]
#     environment:
#       DB: sqlite

# Create namespace (logical isolation — separate task queues, history, rate limits):
temporal operator namespace create --namespace production

# Temporal Cloud connection (Go):
c, _ := client.Dial(client.Options{
    HostPort:  "your-namespace.tmprl.cloud:7233",
    Namespace: "your-namespace",
    ConnectionOptions: client.ConnectionOptions{
        TLS: &tls.Config{},
    },
    Credentials: client.NewAPIKeyStaticCredentials("your-api-key"),
})

// Workflow versioning — safely evolve workflow logic with running executions:
v := workflow.GetVersion(ctx, "payment-v2", workflow.DefaultVersion, 1)
if v == workflow.DefaultVersion {
    workflow.ExecuteActivity(ctx, OldChargeActivity, ...).Get(ctx, nil)
} else {
    workflow.ExecuteActivity(ctx, NewChargeActivity, ...).Get(ctx, nil)
}
// GetVersion replays correctly — old executions use DefaultVersion path,
// new executions use version 1 path. Remove the branch once all old executions complete.

// Testing with testsuite (no Temporal server needed):
suite := testsuite.WorkflowTestSuite{}
env := suite.NewTestWorkflowEnvironment()
env.RegisterActivity(activities)
env.OnActivity(ChargePaymentActivity, mock.Anything, mock.Anything).Return("txn-123", nil)
env.ExecuteWorkflow(OrderWorkflow, testOrder)
suite.True(env.IsWorkflowCompleted())