Creating a Seamless Salesforce Task Pipeline with GCP Pub/Sub and Cloud Run: A Step-by-Step Guide

Introduction

In this post, you'll learn how to connect Google Cloud Pub/Sub → Cloud Run → Salesforce so that when you publish an event to a Pub/Sub topic, a new Task record appears in Salesforce. I’ll cover every step:

• Creating a Pub/Sub topic and push subscription

• Securing access with service accounts and OIDC tokens

• Writing and deploying a Cloud Run service in TypeScript

• Setting up a Salesforce Connected App for Client Credential authentication

By the end, you'll have a fully automated pipeline that you can adapt for Leads, Cases, or any custom object.

Sequence Diagram

sequenceDiagram
    autonumber
    actor Developer
    participant PSTopic    as "Pub/Sub Topic (salesforce_tasks)"
    participant PSSub      as "Pub/Sub Subscription (sf-task-sub)"
    participant CRService  as "Cloud Run Service (sf-task-listener)"
    participant SFAuth     as "Salesforce OAuth2 Connected App"
    participant SFAPI      as "Salesforce REST API (Task)"

    Developer->>PSTopic: Publish {"Subject","WhatId"}
    activate PSTopic
    Note right of PSTopic: Message stored in the topic
    PSTopic->>PSSub: Deliver message
    activate PSSub
    deactivate PSTopic

    Note right of PSSub: Generate OIDC token using cloud-run-pubsub-invoker SA
    PSSub->>CRService: HTTP POST /pubsub/push with OIDC token
    activate CRService
    deactivate PSSub

    Note right of CRService: Verify OIDC token via IAM
    CRService->>CRService: Decode Base64 payload
    Note right of CRService: Payload decoded
    CRService->>SFAuth: POST /services/oauth2/token (client_credential)
    activate SFAuth
    deactivate CRService

    SFAuth-->>CRService: {access_token, instance_url}
    activate CRService
    deactivate SFAuth

    CRService->>SFAPI: POST Task {Subject, WhatId}  Authorization: Bearer <access_token>
    activate SFAPI
    deactivate CRService

    SFAPI-->>CRService: {id: 00TXXXXXXXXXXXX}
    activate CRService
    deactivate SFAPI

    CRService-->>PSSub: HTTP 204 ACK
    activate PSSub
    deactivate PSSub

    CRService->>Developer: Log "Task created: 00T…"
    activate Developer
    deactivate Developer
    deactivate CRService

Prerequisites

Before I begin, make sure you have:

1. brew install --cask google-cloud-sdk

2. gcloud auth login

3. gcloud config set project gcptosalesforcestreaming (where gcptosalesforcestreaming is my project name in Google cloud)

4. In salesforce, I have also setup a Connected App

   1. Create a Connected App in Setup → App Manager → New Connected App

   2. Under API (Enable OAuth Settings):

      1. Callback URL: https://login.salesforce.com/services/oauth2/callback

      2. Scopes: (Use more restrictive scopes in Production, this is just for demo purpose)

         1. Manage user data via APIs (api)

         2. Full access (full)

         3. Perform requests at any time (refresh_token, offline_access)

      3. Enabled “Enable Client Credentials Flow“

      4. Save and note down Consumer Key, Consumer Secret

   3. Click “Manage”

      1. Make Admin approved users are pre-authorized

      2. Add a user under Client Credential Flow

      3. For demo purpose you can use a admin user, but for production usage, follow principle of least privilege

Step 1: Create the Pub/Sub Topic

gcloud pubsub topics create salesforce_tasks

This will be the “event bus” for all Task‑creation requests.

Step 2: Build & Deploy Your Cloud Run Service (TypeScript)

• Scaffold the project

    mkdir cloudrun_sf_task
    cd cloudrun_sf_task
    npm init -y
    npm install express body-parser axios
    npm install --save-dev typescript @types/express @types/node ts-node
  

• Configure TypeScript
  tsconfig.json:

    {
      "compilerOptions": {
        "target": "ES2020",
        "module": "commonjs",
        "outDir": "dist",
        "strict": true,
        "moduleResolution": "node",
        "esModuleInterop": true,
        "allowSyntheticDefaultImports": true
      },
      "include": ["src/**/*.ts"]
    }
  

• Write the handler
  src/index.ts:

    import express from 'express';
    import { json } from 'body-parser';
    import axios from 'axios';
  
    const app = express();
    app.use(json({ limit: '1mb' }));
  
    // Endpoint to receive Pub/Sub push
    // @ts-ignore
    app.post('/pubsub/push', async (req, res) => {
        try {
            // Pub/Sub push envelope
            const message = req.body.message;
            if (!message || !message.data) {
                return res.status(400).send('Invalid Pub/Sub message format');
            }
  
            // Decode the Base64‐encoded message
            const payload = JSON.parse(Buffer.from(message.data, 'base64').toString());
  
            // Extract whatever fields you need, e.g.
            const { Subject, WhatId } = payload;
  
            // Authenticate to Salesforce via OAuth JWT Bearer or client credentials
            const tokenResp = await axios.post(
                `https://${process.env.SF_DOMAIN_URL!}/services/oauth2/token`,
                new URLSearchParams({
                    grant_type: 'client_credentials',
                    client_id: process.env.SF_CLIENT_ID!,
                    client_secret: process.env.SF_CLIENT_SECRET!
                }).toString(),
                { headers: { 'Content-Type': 'application/x-www-form-urlencoded' } }
            );
            const accessToken = tokenResp.data.access_token;
            const instanceUrl = tokenResp.data.instance_url;
  
            // Create the Task
            const taskResp = await axios.post(
                `${instanceUrl}/services/data/v60.0/sobjects/Task`,
                {
                    Subject,
                    WhatId
                },
                {
                    headers: {
                        Authorization: `Bearer ${accessToken}`,
                        'Content-Type': 'application/json'
                    }
                }
            );
  
            console.log('Task created:', taskResp.data.id);
            res.status(204).send();  // 2xx to ack Pub/Sub
        } catch (err: any) {
            console.error('Error processing Pub/Sub:', err.response?.data || err.message);
            res.status(500).send('Internal error');
        }
    });
  
    // health check
    // @ts-ignore
    app.get('/', (_req, res) => res.send('OK'));
  
    const port = parseInt(process.env.PORT || '8080', 10);
    app.listen(port, () => {
        console.log(`Listening on port ${port}`);
    });
  

• Two Stage DockerFile

    # Builder
    FROM node:18-alpine AS builder
    WORKDIR /app
    COPY package*.json tsconfig.json ./
    RUN npm install
    COPY src ./src
    RUN npm run build
  
    # Runtime
    FROM node:18-alpine
    WORKDIR /app
    COPY package*.json ./
    RUN npm install --only=production
    COPY --from=builder /app/dist ./dist
    EXPOSE 8080
    CMD ["node","dist/index.js"]
  

• Build & Push

    npm run build
    gcloud builds submit --tag gcr.io/gcptosalesforcestreaming/sf_task_listener
  

• Deploy with env‑vars

    gcloud run deploy sf-task-listener --image gcr.io/gcptosalesforcestreaming/sf_task_listener --region=us-central1 --allow-unauthenticated --set-env-vars="SF_CLIENT_ID=CLIENT_ID_FROM_CONNECTED_APP,
    SF_CLIENT_SECRET=CLIENT_SECRET_FROM_CONNECTED_APP,SF_DOMAIN_URL=nagesingh-dev-ed.my.salesforce.com"
  

• After successful deployment you should get a “Service URL“

    Service [sf-task-listener] revision [sf-task-listener-00004-zgg] has been deployed and is serving 100 percent of traffic. 
  
    Service URL: https://sf-task-listener-YOUR_PROJECT_NUMBER.us-central1.run.app
  

Step 3: Secure Service Accounts for Pub/Sub → Cloud Run

1. Create an invoker SA

    gcloud iam service-accounts create cloud-run-pubsub-invoker \
      --display-name="Cloud Run Pub/Sub Invoker"
   

2. Allow it to call your Cloud Run service

    gcloud run services add-iam-policy-binding sf-task-listener \
      --region=us-central1 \
      --member="serviceAccount:cloud-run-pubsub-invoker@gcptosalesforcestreaming.iam.gserviceaccount.com" \
      --role="roles/run.invoker"
   

3. Let Pub/Sub mint OIDC tokens

    PROJECT_NUMBER=$(gcloud projects describe $(gcloud config get-value project) --format="value(projectNumber)")
    PUBSUB_SA="service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com"
   
    gcloud iam service-accounts add-iam-policy-binding \
      cloud-run-pubsub-invoker@$(gcloud config get-value project).iam.gserviceaccount.com \
      --member="serviceAccount:${PUBSUB_SA}" \
      --role="roles/iam.serviceAccountTokenCreator"
   

Step 4: Create the Push Subscription

Point your topic at Cloud Run, using your invoker SA for auth:

YOUR_CLOUD_RUN_URL : Step2’s last step.

bashCopyEditgcloud pubsub subscriptions create sf-task-sub \
  --topic=salesforce_tasks \
  --push-endpoint="https://YOUR_CLOUD_RUN_URL/pubsub/push" \
  --push-auth-service-account="cloud-run-pubsub-invoker@gcptosalesforcestreaming.iam.gserviceaccount.com"

Step 5: Test Your Pipeline

Publish a message to the topic salesforce_tasks

WhatId : I have taken a hardcoded AccountId, just for demo purpose.

gcloud pubsub topics publish salesforce_tasks --message='{"Subject":"Follow up call","WhatId":"0017F00002lLmnx"}'

Conclusion

You now have a robust, secure, and fully automated pipeline from GCP Pub/Sub → Cloud Run → Salesforce Tasks. Feel free to adapt this pattern for other Salesforce objects or workflows. Happy building!