import { type ExuluContextSource, type ExuluItem, ExuluQueues } from "@exulu/backend";
import knex, { type Knex } from "knex";
import { createHash } from "crypto";
// ── Connection pool ───────────────────────────────────────────────────────────
// Create the pool at module scope so it is shared across all source executions
// within the same worker process. knex manages idle connections internally.
const externalDb: Knex = knex({
client: "pg",
connection: {
host: process.env.PRODUCT_DB_HOST,
port: Number(process.env.PRODUCT_DB_PORT ?? 5432),
database: process.env.PRODUCT_DB_NAME,
user: process.env.PRODUCT_DB_USER,
password: process.env.PRODUCT_DB_PASSWORD,
ssl: process.env.PRODUCT_DB_SSL === "true" ? { rejectUnauthorized: false } : false,
},
pool: { min: 1, max: 4 }, // keep a small pool; source runs are sequential
});
// ── Row type from the external database ──────────────────────────────────────
type ProductRow = {
id: string;
sku: string;
name: string;
description: string | null;
category: string | null;
price_cents: number;
active: boolean;
updated_at: Date;
};
// ── Item mapping ──────────────────────────────────────────────────────────────
function toExuluItem(row: ProductRow): ExuluItem {
const content_hash = createHash("sha256")
.update(row.name + (row.description ?? "") + String(row.price_cents) + row.category)
.digest("hex");
return {
external_id: row.id,
name: `${row.sku} — ${row.name}`,
body: row.description ?? "",
category: row.category ?? "uncategorised",
price_cents: row.price_cents,
active: row.active,
content_hash,
rights_mode: "users",
};
}
// ── Queue ─────────────────────────────────────────────────────────────────────
export const productSyncQueue = ExuluQueues.register(
"product_catalog_sync",
{ worker: 1, queue: 1 },
1,
60 * 15, // 15-minute timeout
);
// ── Source ────────────────────────────────────────────────────────────────────
export const productCatalogSqlSource: ExuluContextSource = {
id: "product_catalog_sql_source",
name: "Product catalog SQL source",
description: "Reads products updated since the last sync from the external product database.",
config: {
schedule: "0 4 * * *", // daily at 04:00 UTC
queue: productSyncQueue.use(),
retries: 3,
backoff: { type: "exponential", delay: 2_000 },
params: [
{
name: "updated_since",
description: "ISO datetime — fetch rows with updated_at after this value. Defaults to 25 hours ago.",
},
{
name: "full_sync",
description: "Set to 'true' to fetch all rows regardless of updated_at.",
default: "false",
},
],
},
execute: async ({ updated_since: inputUpdatedSince, full_sync: inputFullSync }) => {
const fullSync = inputFullSync === "true";
const updatedSince = fullSync
? undefined
: (inputUpdatedSince ?? new Date(Date.now() - 25 * 60 * 60 * 1_000).toISOString());
console.log("[product_catalog_sql_source] Starting sync", { updatedSince, fullSync });
// ── Query external DB ─────────────────────────────────────────────────────
let query = externalDb<ProductRow>("products")
.select([
"id",
"sku",
"name",
"description",
"category",
"price_cents",
"active",
"updated_at",
])
.orderBy("updated_at", "asc");
if (updatedSince) {
query = query.where("updated_at", ">", new Date(updatedSince));
}
const rows = await query;
console.log(`[product_catalog_sql_source] Fetched ${rows.length} rows`);
// ── Map rows to ExuluItem[] with partial-failure logging ──────────────────
const items: ExuluItem[] = [];
for (const row of rows) {
try {
items.push(toExuluItem(row));
} catch (err) {
console.error(`[product_catalog_sql_source] Failed to map row ${row.id}:`, err);
}
}
console.log(`[product_catalog_sql_source] Returning ${items.length} items`);
return items;
},
};
