deno-pgmq

// Import the Pgmq class from your module
import { Pgmq } from "your-database-module-path"; // Replace with your actual module path

// Load environment variables from a .env file if needed
import "@std/dotenv/load";

console.log("Connecting to Postgres...");
const pgmq = await Pgmq.new({
    dsn: Deno.env.get("DATABASE_URL"), // Use the DSN from the environment variable
    lazy: false, // Set lazy loading based on your preference
    maxPoolSize: Number(Deno.env.get("MAX_POOL_SIZE")) || 20, // Use max pool size from env
}).catch((err) => {
    console.error("Failed to connect to Postgres", err);
    Deno.exit(1);
});

const qName = "my_queue";
console.log(`Creating queue ${qName}...`);
await pgmq.queue.create(qName).catch((err) => {
    console.error("Failed to create queue", err);
    Deno.exit(1);
});

interface Msg {
    id: number;
    name: string;
}
const msg: Msg = { id: 1, name: "testMsg" };
console.log("Sending message...");
const msgId = await pgmq.msg.send(qName, msg).catch((err) => {
    console.error("Failed to send message", err);
    Deno.exit(1);
});

const vt = 30;
const receivedMsg = await pgmq.msg.read<Msg>(qName, vt).catch((err) => {
    console.error("No messages in the queue", err);
    Deno.exit(1);
});

console.log("Received message...");
if (receivedMsg) {
    console.dir(receivedMsg.message, { depth: null });
} else {
    console.log("No message received.");
}

console.log("Archiving message...");
await pgmq.msg.archive(qName, msgId).catch((err) => {
    console.error("Failed to archive message", err);
    Deno.exit(1);
});