Decouple GEO from Exchange

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`.

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" }));
 

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<string, unknown>
+
+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<IngestionResult> {
+  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<IngestionResult> {
+  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<void> {
+  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)
+  })
+}

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<string, ProductCatalogItem>) {
+function productView(doc: ArangoDoc) {
-  const uuid = typeof productUuid === 'string' ? productUuid : ''
-  return uuid ? catalog.get(uuid) : undefined
-}
-
-function productView(doc: ArangoDoc, catalog: Map<string, ProductCatalogItem>) {
   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,
+    name: doc.product_name ?? doc.name ?? null,
-    imageUrl: product?.imageUrl ?? 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<string, ProductCatalogItem>) {
   }
 }
 
-function mapOffer(doc: ArangoDoc, catalog = new Map<string, ProductCatalogItem>()) {
+function mapOffer(doc: ArangoDoc) {
-  const product = catalogProduct(doc.product_uuid, catalog)
   return {
     uuid: doc._key,
     productUuid: doc.product_uuid ?? null,
-    productName: product?.name ?? doc.product_name ?? null,
+    productName: doc.product_name ?? null,
-    productImageUrl: product?.imageUrl ?? 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))
     },
   },
 }