add more connection options config

Author: jsbroks

apps/event-queue/src/config.ts

@@ -21,6 +21,12 @@ export const env = createEnv({
       .refine((arr) => arr.length > 0, {
         message: "KAFKA_BROKERS must be a non-empty list of brokers",
       }),
+
+    KAFKA_PARTITIONS_CONSUMED_CONCURRENTLY: z
+      .number()
+      .default(5)
+      .transform((val) => parseInt(val.toString())),
+
     GITHUB_BOT_APP_ID: z.string().optional(),
     GITHUB_BOT_PRIVATE_KEY: z.string().optional(),
     GITHUB_BOT_CLIENT_ID: z.string().optional(),

apps/event-queue/src/index.ts

@@ -22,6 +22,7 @@ export const start = async () => {
   logger.info("Subscribed to ctrlplane-events topic");
 
   await consumer.run({
+    partitionsConsumedConcurrently: env.KAFKA_PARTITIONS_CONSUMED_CONCURRENTLY,
     eachMessage: async ({ topic, partition, message }) => {
       logger.info("Received event", {
         topic,

packages/db/src/client.ts

@@ -8,8 +8,11 @@ const { Pool } = pg;
 
 export const pool = new Pool({
   max: env.POSTGRES_MAX_POOL_SIZE,
+  min: 1,
   connectionString: env.POSTGRES_URL,
+  keepAlive: true,
   ssl: false,
+  application_name: env.POSTGRES_APPLICATION_NAME,
 });
 
 export const db = drizzle(pool, { schema });

packages/db/src/config.ts

@@ -13,6 +13,8 @@ export const env = createEnv({
       .default("50")
       .transform((value) => parseInt(value)),
 
+    POSTGRES_APPLICATION_NAME: z.string().default("ctrlplane"),
+
     REDIS_URL: z.string().url().default("redis://127.0.0.1:6379"),
   },
   runtimeEnv: process.env,