@blokjs/trigger-pubsub

--- package/template/src/index.ts (excerpt) ---
import { DefaultLogger } from "@blokjs/runner";
import { type Span, metrics, trace } from "@opentelemetry/api";
import PubSubServer from "./runner/PubSubServer";

export default class App {
	private pubsubServer: PubSubServer = <PubSubServer>{};
	protected trigger_initializer = 0;
	protected initializer = 0;
	protected tracer = trace.getTracer(
		process.env.PROJECT_NAME || "trigger-pubsub-server",
		process.env.PROJECT_VERSION || "0.0.1",
	);
	private logger = new DefaultLogger();
	protected app_cold_start = metrics.getMeter("default").createGauge("initialization", {
		description: "Application cold start",
	});

	constructor() {
		this.initializer = performance.now();
		this.pubsubServer = new PubSubServer();
	}

	async run() {
		this.tracer.startActiveSpan("initialization", async (span: Span) => {
			await this.pubsubServer.listen();
			this.initializer = performance.now() - this.initializer;

			this.logger.log(`Pub/Sub trigger initialized in ${this.initializer.toFixed(2)}ms`);
			this.app_cold_start.record(this.initializer, {
				pid: process.pid,
				env: process.env.NODE_ENV,
				app: process.env.APP_NAME,
			});
			span.end();
		});
	}
}

if (process.env.DISABLE_TRIGGER_RUN !== "true") {
	new App().run();
}


--- package/template/src/workflows/messages/on-message.ts (excerpt) ---
import { workflow } from "@blokjs/helper";

/**
 * Example Pub/Sub workflow — fires when a message arrives on a subscription.
 *
 * Message payload + metadata on ctx.request:
 *   - ctx.request.body                  — the message payload
 *   - ctx.request.headers               — message attributes
 *   - ctx.request.params.topic          — topic name
 *   - ctx.request.params.subscription   — subscription name
 *   - ctx.request.params.messageId      — unique message ID
 *   - ctx.vars._pubsub_message          — full broker metadata
 *
 * Pick a provider in the trigger config:
 *   provider: "gcp" | "aws" | "azure"
 *
 * v2 reliability knobs available on each step (uncomment to use):
 *   idempotencyKey: "$.req.params.messageId" — at-most-once delivery semantics
 *   retry: { maxAttempts: 3 }                 — retry on transient failures
 */
export default workflow({
	name: "On Pub/Sub Message",
	version: "1.0.0",
	description: "Handles incoming Pub/Sub messages",
	trigger: {
		pubsub: {
			provider: "gcp",
			topic: "my-topic",
			subscription: "my-subscription",
		},
	},
	steps: [
		{
			id: "log-message",
			use: "@blokjs/api-call",
			type: "module",
			inputs: {
				url: "https://httpbin.org/post",
				method: "POST",
				body: {
					message: "js/ctx.request.body",
					topic: "js/ctx.request.params.topic",
					messageId: "js/ctx.request.params.messageId",
				},
			},
		},
	],
});


--- package/template/src/runner/PubSubServer.ts (excerpt) ---
import { GCPPubSubAdapter, PubSubTrigger } from "@blokjs/trigger-pubsub";
import nodes from "../Nodes";
import workflows from "../Workflows";

/**
 * PubSubServer - Concrete Pub/Sub trigger implementation
 *
 * This server extends the abstract PubSubTrigger and provides:
 * - A specific adapter (GCP Pub/Sub by default, can be changed to AWS or Azure)
 * - Node and workflow registries
 *
 * To change the provider, replace:
 * - GCPPubSubAdapter with AWSSNSAdapter or AzureServiceBusAdapter
 * - Update the adapter configuration accordingly
 *
 * @example AWS SNS/SQS
 * ```typescript
 * import { AWSSNSAdapter } from "@blokjs/trigger-pubsub";
 * protected adapter = new AWSSNSAdapter({
 *   region: process.env.AWS_REGION || "us-east-1",
 * });
 * ```
 *
 * @example Azure Service Bus
 * ```typescript
 * import { AzureServiceBusAdapter } from "@blokjs/trigger-pubsub";
 * protected adapter = new AzureServiceBusAdapter({
 *   connectionString: process.env.AZURE_SERVICE_BUS_CONNECTION_STRING || "",
 * });
 * ```
 */
export default class PubSubServer extends PubSubTrigger {
	protected adapter = new GCPPubSubAdapter({
		projectId: process.env.GCP_PROJECT_ID || "my-project",
	});

	protected nodes: Record<string, import("@blokjs/runner").BlokService<unknown>> = nodes;
	protected workflows: Record<string, import("@blokjs/helper").HelperResponse> = workflows;
}