added cron scheduler

services/embedding-indexer/package.json

@@ -20,6 +20,7 @@
   },
   "dependencies": {
     "axios": "^1.6.7",
+    "cron": "^3.1.6",
     "node-cron": "^3.0.3",
     "pg": "^8.11.3"
   }

services/embedding-indexer/src/app.ts

@@ -1,4 +1,9 @@
-import cron from "node-cron";
 import { reindex } from "./reindex";
+import { CronJob } from "cron";
 
-reindex()
+const job = CronJob.from({
+  cronTime: "0 */15 * * * *",
+  onTick: reindex,
+  start: true,
+  runOnInit: true,
+});

services/embedding-indexer/src/db.ts

@@ -8,9 +8,9 @@ export const query = async (text: string, params: Array<any>) => {
   const start = performance.now();
   const res = await pool.query(text, params);
   const duration = Math.trunc(performance.now() - start);
-  console.log(
-    `Executed query: { rows: ${res.rowCount}, duration: ${duration}} `
-  );
+  // console.log(
+  //   `Executed query: { rows: ${res.rowCount}, duration: ${duration}} `
+  // );
   return res;
 };
 

services/embedding-indexer/src/reindex.ts

@@ -2,19 +2,19 @@ import { query } from "./db";
 import Axios from "axios";
 
 export const reindex = async () => {
+  console.log("Starting embedding reindex");
   const results = await query(
-    // TODO: Remove limit
     `
     SELECT id
     FROM image
     WHERE
       embedding IS NULL
       AND NOT index_error
-    LIMIT 1
   `,
     []
   );
-  results.rows.map((row) => indexItem(row.id));
+  await Promise.all(results.rows.map((row) => indexItem(row.id)));
+  console.log("Completed embedding index");
 };
 
 const indexItem = async (id: string) => {
@@ -26,15 +26,15 @@ const indexItem = async (id: string) => {
   )}`;
   const embeddingInsertQuery = `
   UPDATE image
-  SET test_embedding = $1::vector
+  SET embedding = $1::vector
   WHERE id = $2
   `;
   const indexErrorQuery = `
   UPDATE image
   SET index_error = true
   WHERE id = $1
   `;
-  Axios.get(clip_url)
+  return Axios.get(clip_url)
     .then((result) => JSON.stringify(result.data))
     .then((embedding) => {
       query(embeddingInsertQuery, [embedding, id]);