From 34e56080b6786d5a64935a725500c461d0b17620 Mon Sep 17 00:00:00 2001 From: Ruslan Bakiev <572431+veikab@users.noreply.github.com> Date: Mon, 28 Sep 2026 12:29:24 +0700 Subject: [PATCH] Decouple GEO from Exchange --- README.md | 14 +++ src/index.ts | 4 + src/ingest.ts | 264 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/schema.ts | 91 ++++------------- 4 files changed, 299 insertions(+), 74 deletions(-) create mode 100644 README.md create mode 100644 src/ingest.ts diff --git a/README.md b/README.md new file mode 100644 index 0000000..6f75194 --- /dev/null +++ b/README.md @@ -0,0 +1,14 @@ +# Optovia GEO + +GEO owns the ArangoDB search graph and public map queries. It does not own +suppliers, offers, orders, logistics drafts or billing state. + +Kestra writes projections through two internal endpoints protected by +`GEO_INGESTION_TOKEN`: + +- `POST /internal/ingest/catalog` replaces the current Odoo supplier/offer projection. +- `POST /internal/ingest/reference` replaces one named source of reference GEO nodes. + +Both endpoints accept at most 1,000 records per collection and return the +three ingestion counters used by Kestra: `loaded`, `loadedWithWarnings` and +`notLoaded`. diff --git a/src/index.ts b/src/index.ts index 0c47777..ca566f6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,10 +2,13 @@ import Fastify from "fastify"; import cors from "@fastify/cors"; import mercurius from "mercurius"; import { typeDefs, resolvers } from "./schema.js"; +import { registerIngestionRoutes } from "./ingest.js"; +import { ensureGraph } from "./db.js"; const PORT = Number.parseInt(process.env.PORT || "8000", 10); const app = Fastify(); +await ensureGraph(); await app.register(cors, { origin: ["https://optovia.ru"], credentials: true }); await app.register(mercurius, { schema: typeDefs, @@ -13,6 +16,7 @@ await app.register(mercurius, { path: "/graphql/public", graphiql: true, }); +await registerIngestionRoutes(app); app.get("/health", async () => ({ status: "ok" })); diff --git a/src/ingest.ts b/src/ingest.ts new file mode 100644 index 0000000..c696b76 --- /dev/null +++ b/src/ingest.ts @@ -0,0 +1,264 @@ +import { timingSafeEqual } from 'node:crypto' +import type { FastifyInstance, FastifyRequest } from 'fastify' +import { getDb } from './db.js' + +const MAX_BATCH_SIZE = 1000 + +type IngestionResult = { + loaded: number + loadedWithWarnings: number + notLoaded: number + warnings: string[] + errors: string[] +} + +type CatalogPayload = { + suppliers?: unknown + offers?: unknown + truncated?: unknown +} + +type ReferencePayload = { + source?: unknown + nodes?: unknown +} + +type JsonRecord = Record + +function secureTokenMatch(supplied: string, configured: string): boolean { + const suppliedBuffer = Buffer.from(supplied) + const configuredBuffer = Buffer.from(configured) + return suppliedBuffer.length === configuredBuffer.length + && timingSafeEqual(suppliedBuffer, configuredBuffer) +} + +function requireIngestionToken(request: FastifyRequest, configuredToken: string): void { + const suppliedToken = request.headers['x-ingestion-token'] + if (typeof suppliedToken !== 'string' || !secureTokenMatch(suppliedToken, configuredToken)) { + const error = new Error('Unauthorized GEO ingestion request') as Error & { statusCode: number } + error.statusCode = 401 + throw error + } +} + +function asRecords(value: unknown, field: string): JsonRecord[] { + if (!Array.isArray(value)) throw new Error(`${field} must be an array`) + if (value.length > MAX_BATCH_SIZE) throw new Error(`${field} exceeds ${MAX_BATCH_SIZE} records`) + return value.map((item, index) => { + if (item === null || typeof item !== 'object' || Array.isArray(item)) { + throw new Error(`${field}[${index}] must be an object`) + } + return item as JsonRecord + }) +} + +function requiredString(record: JsonRecord, key: string): string { + const value = record[key] + if (typeof value !== 'string' || value.trim().length === 0) { + throw new Error(`${key} is required`) + } + return value.trim() +} + +function optionalString(record: JsonRecord, key: string): string | null { + const value = record[key] + if (value == null || value === '') return null + if (typeof value !== 'string') throw new Error(`${key} must be a string`) + return value.trim() +} + +function requiredCoordinate(record: JsonRecord, key: string, min: number, max: number): number { + const value = record[key] + if (typeof value !== 'number' || !Number.isFinite(value) || value < min || value > max) { + throw new Error(`${key} must be between ${min} and ${max}`) + } + return value +} + +function optionalNumber(record: JsonRecord, key: string): number | null { + const value = record[key] + if (value == null || value === '') return null + const parsed = typeof value === 'number' ? value : Number(value) + if (!Number.isFinite(parsed)) throw new Error(`${key} must be numeric`) + return parsed +} + +async function upsertCatalog(payload: CatalogPayload): Promise { + if (payload.truncated === true) throw new Error('Odoo GEO export was truncated') + + const supplierRecords = asRecords(payload.suppliers ?? [], 'suppliers') + const offerRecords = asRecords(payload.offers ?? [], 'offers') + const warnings: string[] = [] + let loaded = 0 + let loadedWithWarnings = 0 + + const suppliers = supplierRecords + .map((item, index) => { + const countryCode = optionalString(item, 'countryCode') + if (!countryCode) { + warnings.push(`suppliers[${index}]: countryCode is missing`) + loadedWithWarnings += 1 + } else { + loaded += 1 + } + return { + _key: requiredString(item, 'uuid'), + name: requiredString(item, 'name'), + country: optionalString(item, 'country'), + country_code: countryCode, + latitude: requiredCoordinate(item, 'latitude', -90, 90), + longitude: requiredCoordinate(item, 'longitude', -180, 180), + node_type: 'supplier', + source: 'odoo', + } + }) + + const offers = offerRecords + .map((item, index) => { + const countryCode = optionalString(item, 'countryCode') + if (!countryCode) { + warnings.push(`offers[${index}]: countryCode is missing`) + loadedWithWarnings += 1 + } else { + loaded += 1 + } + return { + _key: requiredString(item, 'uuid'), + name: requiredString(item, 'productName'), + product_uuid: requiredString(item, 'productUuid'), + product_name: requiredString(item, 'productName'), + product_image_url: optionalString(item, 'productImageUrl'), + supplier_uuid: requiredString(item, 'supplierUuid'), + supplier_name: requiredString(item, 'supplierName'), + supplier_logo_url: optionalString(item, 'supplierLogoUrl'), + latitude: requiredCoordinate(item, 'latitude', -90, 90), + longitude: requiredCoordinate(item, 'longitude', -180, 180), + country: optionalString(item, 'country'), + country_code: countryCode, + price_per_unit: optionalString(item, 'pricePerUnit'), + currency: optionalString(item, 'currency'), + quantity: optionalString(item, 'quantity'), + unit: optionalString(item, 'unit'), + node_type: 'offer', + source: 'odoo', + } + }) + + const documents = [...suppliers, ...offers] + const keys = documents.map((document) => document._key) + const db = getDb() + const cursor = await db.query(` + FOR document IN @documents + UPSERT { _key: document._key } + INSERT MERGE(document, { synced_at: DATE_ISO8601(DATE_NOW()) }) + UPDATE MERGE(document, { synced_at: DATE_ISO8601(DATE_NOW()) }) + IN nodes + COLLECT WITH COUNT INTO count + RETURN count + `, { documents }) + await cursor.all() + + const cleanupCursor = await db.query(` + FOR document IN nodes + FILTER document.source == 'odoo' + FILTER document.node_type == 'supplier' OR document.node_type == 'offer' + FILTER document._key NOT IN @keys + REMOVE document IN nodes + COLLECT WITH COUNT INTO count + RETURN count + `, { keys }) + await cleanupCursor.all() + + return { + loaded, + loadedWithWarnings, + notLoaded: 0, + warnings, + errors: [], + } +} + +async function upsertReference(payload: ReferencePayload): Promise { + const source = typeof payload.source === 'string' && payload.source.trim() + ? payload.source.trim() + : null + if (!source) throw new Error('source is required') + + const nodeRecords = asRecords(payload.nodes ?? [], 'nodes') + const warnings: string[] = [] + let loaded = 0 + let loadedWithWarnings = 0 + + const nodes = nodeRecords + .map((item, index) => { + const transportTypes = item.transportTypes + if (!Array.isArray(transportTypes) || transportTypes.length === 0) { + warnings.push(`nodes[${index}]: transportTypes is empty`) + loadedWithWarnings += 1 + } else { + loaded += 1 + } + return { + _key: requiredString(item, 'uuid'), + name: requiredString(item, 'name'), + code: optionalString(item, 'code'), + country: optionalString(item, 'country'), + country_code: optionalString(item, 'countryCode'), + latitude: requiredCoordinate(item, 'latitude', -90, 90), + longitude: requiredCoordinate(item, 'longitude', -180, 180), + transport_types: Array.isArray(transportTypes) + ? transportTypes.map((value) => String(value)) + : [], + navigation_priority: optionalNumber(item, 'navigationPriority') ?? 0, + node_type: 'logistics', + source, + } + }) + + const keys = nodes.map((node) => node._key) + const db = getDb() + const cursor = await db.query(` + FOR document IN @documents + UPSERT { _key: document._key } + INSERT MERGE(document, { synced_at: DATE_ISO8601(DATE_NOW()) }) + UPDATE MERGE(document, { synced_at: DATE_ISO8601(DATE_NOW()) }) + IN nodes + COLLECT WITH COUNT INTO count + RETURN count + `, { documents: nodes }) + await cursor.all() + + const cleanupCursor = await db.query(` + FOR document IN nodes + FILTER document.source == @source + FILTER document.node_type == 'logistics' + FILTER document._key NOT IN @keys + REMOVE document IN nodes + COLLECT WITH COUNT INTO count + RETURN count + `, { source, keys }) + await cleanupCursor.all() + + return { + loaded, + loadedWithWarnings, + notLoaded: 0, + warnings, + errors: [], + } +} + +export async function registerIngestionRoutes(app: FastifyInstance): Promise { + const configuredToken = process.env.GEO_INGESTION_TOKEN + if (!configuredToken) throw new Error('GEO_INGESTION_TOKEN is required') + + app.post('/internal/ingest/catalog', async (request) => { + requireIngestionToken(request, configuredToken) + return upsertCatalog(request.body as CatalogPayload) + }) + + app.post('/internal/ingest/reference', async (request) => { + requireIngestionToken(request, configuredToken) + return upsertReference(request.body as ReferencePayload) + }) +} diff --git a/src/schema.ts b/src/schema.ts index b131043..e74ff12 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -4,22 +4,8 @@ import { distanceKm, buildRouteFromEdges, type ArangoDoc, type RoutePath } from const GRAPHHOPPER_URL = process.env.GRAPHHOPPER_EXTERNAL_URL || 'https://graphhopper.optovia.ru' const OPENRAILROUTING_URL = process.env.OPENRAILROUTING_EXTERNAL_URL || 'https://openrailrouting.optovia.ru' -const EXCHANGE_GRAPHQL_URL = process.env.EXCHANGE_GRAPHQL_URL || 'https://exchange.optovia.ru/graphql' const MAX_EXPANSIONS = 20000 -type ProductCatalogItem = { - uuid: string - name: string | null - imageUrl: string | null -} - -type ProductCatalogPayload = { - data?: { - products?: ProductCatalogItem[] - } - errors?: { message?: string }[] -} - export const typeDefs = `#graphql type Edge { toUuid: String @@ -247,18 +233,12 @@ function mapNode(doc: ArangoDoc, includeEdges = false) { } } -function catalogProduct(productUuid: unknown, catalog: Map) { - const uuid = typeof productUuid === 'string' ? productUuid : '' - return uuid ? catalog.get(uuid) : undefined -} - -function productView(doc: ArangoDoc, catalog: Map) { +function productView(doc: ArangoDoc) { const productUuid = doc.product_uuid ?? doc.uuid ?? null - const product = catalogProduct(productUuid, catalog) return { uuid: productUuid, - name: product?.name ?? doc.product_name ?? doc.name ?? null, - imageUrl: product?.imageUrl ?? null, + name: doc.product_name ?? doc.name ?? null, + imageUrl: doc.product_image_url ?? null, offersCount: doc.offers_count ?? doc.offersCount ?? null, averagePrice: doc.average_price ?? null, priceVariationPercent: doc.price_variation_percent ?? null, @@ -268,13 +248,12 @@ function productView(doc: ArangoDoc, catalog: Map) { } } -function mapOffer(doc: ArangoDoc, catalog = new Map()) { - const product = catalogProduct(doc.product_uuid, catalog) +function mapOffer(doc: ArangoDoc) { return { uuid: doc._key, productUuid: doc.product_uuid ?? null, - productName: product?.name ?? doc.product_name ?? null, - productImageUrl: product?.imageUrl ?? null, + productName: doc.product_name ?? null, + productImageUrl: doc.product_image_url ?? null, supplierUuid: doc.supplier_uuid ?? null, supplierName: doc.supplier_name ?? null, supplierLogoUrl: doc.supplier_logo_url ?? null, @@ -469,36 +448,6 @@ async function resolveLandingNetworkStory(focusKind: LandingNetworkKind, itemLim } } -async function loadProductCatalog() { - const res = await fetch(EXCHANGE_GRAPHQL_URL, { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify({ - query: ` - query GeoProductCatalog { - products(limit: 1000) { - uuid - name - imageUrl - } - } - `, - }), - }) - if (!res.ok) throw new Error(`Exchange product catalog request failed: ${res.status}`) - - const payload = await res.json() as ProductCatalogPayload - if (payload.errors?.length) { - throw new Error(payload.errors.map((error) => error.message ?? 'Exchange GraphQL error').join('\n')) - } - - return new Map( - (payload.data?.products ?? []) - .filter((product) => product.uuid) - .map((product) => [product.uuid, product]), - ) -} - function mapEdge(doc: ArangoDoc) { return { toUuid: doc.to_uuid ?? doc._key ?? null, @@ -911,22 +860,20 @@ export const resolvers = { products: async () => { const db = getDb() - const productCatalog = await loadProductCatalog() const cursor = await db.query(` FOR node IN nodes FILTER node.node_type == 'offer' FILTER node.product_uuid != null COLLECT product_uuid = node.product_uuid INTO offers LET first_offer = FIRST(offers).node - RETURN { uuid: product_uuid, product_uuid, product_name: first_offer.product_name, offers_count: LENGTH(offers) } + RETURN { uuid: product_uuid, product_uuid, product_name: first_offer.product_name, product_image_url: first_offer.product_image_url, offers_count: LENGTH(offers) } `) const products = await cursor.all() - return products.map((product: ArangoDoc) => productView(product, productCatalog)) + return products.map((product: ArangoDoc) => productView(product)) }, offersByProduct: async (_: unknown, args: { productUuid: string }) => { const db = getDb() - const productCatalog = await loadProductCatalog() const cursor = await db.query(` FOR node IN nodes FILTER node.node_type == 'offer' @@ -934,7 +881,7 @@ export const resolvers = { RETURN node `, { product_uuid: args.productUuid }) const nodes = await cursor.all() - return nodes.map((node: ArangoDoc) => mapOffer(node, productCatalog)) + return nodes.map((node: ArangoDoc) => mapOffer(node)) }, hubsNearOffer: async (_: unknown, args: { offerUuid: string; limit?: number }) => { @@ -980,7 +927,6 @@ export const resolvers = { productsBySupplier: async (_: unknown, args: { supplierUuid: string }) => { const db = getDb() - const productCatalog = await loadProductCatalog() const cursor = await db.query(` FOR node IN nodes FILTER node.node_type == 'offer' @@ -988,15 +934,14 @@ export const resolvers = { FILTER node.product_uuid != null COLLECT product_uuid = node.product_uuid INTO offers LET first_offer = FIRST(offers).node - RETURN { uuid: product_uuid, product_uuid, product_name: first_offer.product_name, offers_count: LENGTH(offers) } + RETURN { uuid: product_uuid, product_uuid, product_name: first_offer.product_name, product_image_url: first_offer.product_image_url, offers_count: LENGTH(offers) } `, { supplier_uuid: args.supplierUuid }) const products = await cursor.all() - return products.map((product: ArangoDoc) => productView(product, productCatalog)) + return products.map((product: ArangoDoc) => productView(product)) }, offersBySupplierProduct: async (_: unknown, args: { supplierUuid: string; productUuid: string }) => { const db = getDb() - const productCatalog = await loadProductCatalog() const cursor = await db.query(` FOR node IN nodes FILTER node.node_type == 'offer' @@ -1005,12 +950,11 @@ export const resolvers = { RETURN node `, { supplier_uuid: args.supplierUuid, product_uuid: args.productUuid }) const nodes = await cursor.all() - return nodes.map((node: ArangoDoc) => mapOffer(node, productCatalog)) + return nodes.map((node: ArangoDoc) => mapOffer(node)) }, productsNearHub: async (_: unknown, args: { hubUuid: string; radiusKm?: number }) => { const db = getDb() - const productCatalog = await loadProductCatalog() const nodesCol = db.collection('nodes') const hub = await nodesCol.document(args.hubUuid).catch(() => null) if (!hub || hub.latitude == null || hub.longitude == null) return [] @@ -1024,10 +968,10 @@ export const resolvers = { FILTER dist <= @radius_km COLLECT product_uuid = node.product_uuid INTO offers LET first_offer = FIRST(offers).node - RETURN { uuid: product_uuid, product_uuid, product_name: first_offer.product_name, offers_count: LENGTH(offers) } + RETURN { uuid: product_uuid, product_uuid, product_name: first_offer.product_name, product_image_url: first_offer.product_image_url, offers_count: LENGTH(offers) } `, { lat: hub.latitude, lon: hub.longitude, radius_km: args.radiusKm ?? 500 }) const products = await cursor.all() - return products.map((product: ArangoDoc) => productView(product, productCatalog)) + return products.map((product: ArangoDoc) => productView(product)) }, suppliersForProduct: async (_: unknown, args: { productUuid: string }) => { @@ -1219,7 +1163,6 @@ export const resolvers = { nearestOffers: async (_: unknown, args: { lat: number; lon: number; radius?: number; productUuid?: string; hubUuid?: string; limit?: number }) => { const db = getDb() - const productCatalog = await loadProductCatalog() await ensureGraph() const radius = args.radius ?? 500 const limit = args.limit ?? 50 @@ -1252,7 +1195,7 @@ export const resolvers = { const routeResult = await resolveOfferToHubInternal(node._key, args.hubUuid) if (routeResult?.routes) routes = routeResult.routes as RoutePath[] } - offers.push({ ...mapOffer(node, productCatalog), routes }) + offers.push({ ...mapOffer(node), routes }) } return offers }, @@ -1368,7 +1311,6 @@ export const resolvers = { productsList: async (_: unknown, args: { limit?: number; offset?: number; west?: number; south?: number; east?: number; north?: number }) => { const db = getDb() - const productCatalog = await loadProductCatalog() const bounds = boundsFilter(args) const cursor = await db.query(` @@ -1395,6 +1337,7 @@ export const resolvers = { uuid: product_uuid, product_uuid, product_name: first_offer.product_name, + product_image_url: first_offer.product_image_url, offers_count: LENGTH(offers), average_price, price_variation_percent, @@ -1404,7 +1347,7 @@ export const resolvers = { } `, { offset: args.offset ?? 0, limit: args.limit ?? 50, ...bounds.vars }) const products = await cursor.all() - return products.map((product: ArangoDoc) => productView(product, productCatalog)) + return products.map((product: ArangoDoc) => productView(product)) }, }, }