.

config/deno-kv.ts

@@ -1,10 +1,15 @@
 import type { PoolClient, Transaction } from "postgres";
 import { dbPool } from "./postgres.ts";
+import { openaiAssistant } from "../services/openai-assistant.ts";
 
 let db: Deno.Kv | null = null;
 
 export interface KvMessage {
-  type: "CREATE_LISTING" | "CREATE_RAW_LAMUDI_DATA" | "PROPERTY_VALUATION";
+  type:
+    | "CREATE_LISTING"
+    | "CREATE_RAW_LAMUDI_DATA"
+    | "CREATE_AI_GENERATED_DESCRIPTION"
+    | "PROPERTY_VALUATION";
   source: "LAMUDI" | "APP";
   // deno-lint-ignore no-explicit-any
   data: any;
@@ -441,9 +446,60 @@ export async function listenQueue(kv: Deno.Kv) {
           }
         }
         break;
-      case "PROPERTY_VALUATION":
-        if (msg.source === "APP") {
-          console.log(JSON.stringify(msg, null, 2));
+      case "CREATE_AI_GENERATED_DESCRIPTION":
+        {
+          let transaction: Transaction | null = null;
+          const client_1 = await dbPool.connect();
+          try {
+            transaction = client_1.createTransaction(
+              "create-ai-generated-description"
+            );
+
+            await transaction.begin();
+
+            const property = await transaction.queryObject({
+              args: [],
+              text: `SELECT * FROM Property WHERE ai_generated_description IS NULL LIMIT 10 ORDER BY created_at DESC`,
+            });
+
+            if (property.rowCount && property.rowCount > 0 && transaction) {
+              // Process properties in parallel with rate limiting
+              const processProperty = async (row: unknown) => {
+                const propertyData = row as {
+                  id: number;
+                };
+
+                const aiGeneratedDescription = await openaiAssistant(
+                  JSON.stringify(row)
+                );
+
+                if (aiGeneratedDescription) {
+                  await transaction.queryObject({
+                    args: [aiGeneratedDescription, propertyData.id],
+                    text: `UPDATE Property SET ai_generated_description = $1 WHERE id = $2`,
+                  });
+                }
+              };
+
+              // Process 2 properties at a time with 5s delay between batches
+              for (let i = 0; i < property.rows.length; i += 2) {
+                const batch = property.rows.slice(i, i + 2);
+                await Promise.all(batch.map(processProperty));
+                if (i + 2 < property.rows.length) {
+                  await new Promise((resolve) => setTimeout(resolve, 5000));
+                }
+              }
+            }
+
+            await transaction.commit();
+            console.log("Transaction successfully committed for create");
+          } catch (error) {
+            if (transaction) await transaction.rollback();
+            console.error(error);
+          } finally {
+            client_1.release();
+            console.log("Connection released");
+          }
         }
         break;
     }

deno.json

@@ -6,6 +6,7 @@
     "@hono/clerk-auth": "npm:@hono/clerk-auth@^2.0.0",
     "@std/dotenv": "jsr:@std/dotenv@^0.225.2",
     "hono": "npm:hono@^4.6.6",
+    "openai": "npm:openai@^4.68.4",
     "postgres": "https://deno.land/x/[email protected]/mod.ts",
     "zod": "npm:zod@^3.23.8"
   },

schema.sql

@@ -71,7 +71,7 @@ CREATE TABLE Property (
     property_features JSONB,
     indoor_features JSONB,
     outdoor_features JSONB,
-    ai_generated_description TEXT,
+    ai_generated_description JSONB,
     ai_generated_basic_features JSONB,
     property_type_id INT NOT NULL REFERENCES Property_Type(property_type_id),
     warehouse_type_id INT REFERENCES Warehouse_Type(warehouse_type_id),

server.ts

@@ -374,6 +374,7 @@ app.get("/api/properties/:id", async (c: Context) => {
 
 app.post("/", async (c: Context) => {
   const data = await c.req.json();
+  console.log(JSON.stringify(data));
   await sendMessage({ kv, data, options: { delay: 5000 } });
   return c.text("Hono!");
 });