feat: add server-side ETL pipeline, revert client to NocoDB reads
ETL Pipeline (server): - POST /api/etl/sync?mode=full|incremental — fetches ERP, aggregates, writes NocoDB - nocodbClient.ts: table discovery, paginated delete/insert - etlSync.ts: orchestrates fetch → aggregate → upsert - museumMapping.ts moved from client to server - Auth via ETL_SECRET bearer token Client: - dataService.ts reverts to reading NocoDB DailySales table - Paginated fetch via fetchNocoDBTable (handles >1000 rows) - Suspicious data check: prefers cache if NocoDB returns <10 rows - Deleted erpService.ts and client-side museumMapping.ts First full sync: 391K transactions → 5,760 daily records in 108s. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -7,11 +7,10 @@ ERP_API_CODE=your-api-function-key
|
||||
ERP_USERNAME=your-username
|
||||
ERP_PASSWORD=your-password
|
||||
|
||||
# Salla OAuth Credentials (from Salla Partners dashboard)
|
||||
SALLA_CLIENT_ID=your_client_id_here
|
||||
SALLA_CLIENT_SECRET=your_client_secret_here
|
||||
SALLA_REDIRECT_URI=http://localhost:3001/auth/callback
|
||||
# NocoDB (for ETL writes)
|
||||
NOCODB_URL=http://localhost:8090
|
||||
NOCODB_TOKEN=your-token
|
||||
NOCODB_BASE_ID=your-base-id
|
||||
|
||||
# After OAuth, these will be populated automatically
|
||||
# SALLA_ACCESS_TOKEN=
|
||||
# SALLA_REFRESH_TOKEN=
|
||||
# ETL sync secret (for cron auth)
|
||||
ETL_SECRET=your-secret-here
|
||||
|
||||
@@ -23,3 +23,13 @@ export const erp = {
|
||||
username: process.env.ERP_USERNAME || '',
|
||||
password: process.env.ERP_PASSWORD || '',
|
||||
};
|
||||
|
||||
export const nocodb = {
|
||||
url: process.env.NOCODB_URL || '',
|
||||
token: process.env.NOCODB_TOKEN || '',
|
||||
baseId: process.env.NOCODB_BASE_ID || '',
|
||||
};
|
||||
|
||||
export const etl = {
|
||||
secret: process.env.ETL_SECRET || '',
|
||||
};
|
||||
|
||||
38
server/src/config/museumMapping.ts
Normal file
38
server/src/config/museumMapping.ts
Normal file
@@ -0,0 +1,38 @@
|
||||
// Definitive mapping of ERP product descriptions to museum names.
|
||||
// Priority order matters — first match wins (handles combo tickets).
|
||||
|
||||
const MUSEUM_KEYWORDS: [string, string[]][] = [
|
||||
['Revelation Exhibition', ['Revelation', 'الوحي']],
|
||||
['Creation Story Museum', ['Creation Story', 'قصة الخلق']],
|
||||
['Holy Quraan Museum', ['Holy Quraan', 'القرآن الكريم']],
|
||||
['Trail To Hira Cave', ['Trail To Hira', 'غار حراء']],
|
||||
['Makkah Greets Us', ['Makkah Greets']],
|
||||
['VIP Experience', ['VIP Experience']],
|
||||
];
|
||||
|
||||
export const MUSEUM_NAMES = MUSEUM_KEYWORDS.map(([name]) => name);
|
||||
|
||||
export function getMuseumFromProduct(productDescription: string): string {
|
||||
const desc = productDescription.trim();
|
||||
for (const [museum, keywords] of MUSEUM_KEYWORDS) {
|
||||
for (const kw of keywords) {
|
||||
if (desc.includes(kw)) return museum;
|
||||
}
|
||||
}
|
||||
return 'Other';
|
||||
}
|
||||
|
||||
export const CHANNEL_LABELS: Record<string, string> = {
|
||||
'B2C': 'HiHala Website/App',
|
||||
'B2B': 'B2B',
|
||||
'POS': 'POS',
|
||||
'Safiyyah POS': 'Safiyyah POS',
|
||||
'Standalone': 'Standalone',
|
||||
'Mobile': 'Mobile',
|
||||
'Viva': 'Viva',
|
||||
'IT': 'IT',
|
||||
};
|
||||
|
||||
export function getChannelLabel(operatingAreaName: string): string {
|
||||
return CHANNEL_LABELS[operatingAreaName] || operatingAreaName;
|
||||
}
|
||||
@@ -1,7 +1,8 @@
|
||||
import express from 'express';
|
||||
import cors from 'cors';
|
||||
import { server, erp } from './config';
|
||||
import { server, erp, nocodb } from './config';
|
||||
import erpRoutes from './routes/erp';
|
||||
import etlRoutes from './routes/etl';
|
||||
|
||||
const app = express();
|
||||
app.use(cors());
|
||||
@@ -9,14 +10,21 @@ app.use(express.json());
|
||||
|
||||
// Mount routes
|
||||
app.use('/api/erp', erpRoutes);
|
||||
app.use('/api/etl', etlRoutes);
|
||||
|
||||
app.listen(server.port, () => {
|
||||
console.log(`\nServer running on http://localhost:${server.port}`);
|
||||
|
||||
if (erp.apiUrl && erp.username) {
|
||||
console.log(' GET /api/erp/sales?startDate=...&endDate=...');
|
||||
console.log(' GET /api/erp/status');
|
||||
console.log(' ERP: configured');
|
||||
} else {
|
||||
console.log(' WARNING: ERP_API_URL / ERP_USERNAME not set in .env');
|
||||
console.log(' ERP: WARNING — not configured');
|
||||
}
|
||||
|
||||
if (nocodb.url && nocodb.token) {
|
||||
console.log(' NocoDB: configured');
|
||||
console.log(' POST /api/etl/sync?mode=full|incremental');
|
||||
} else {
|
||||
console.log(' NocoDB: WARNING — not configured');
|
||||
}
|
||||
});
|
||||
|
||||
34
server/src/routes/etl.ts
Normal file
34
server/src/routes/etl.ts
Normal file
@@ -0,0 +1,34 @@
|
||||
import { Router, Request, Response } from 'express';
|
||||
import { etl } from '../config';
|
||||
import { runSync } from '../services/etlSync';
|
||||
|
||||
const router = Router();
|
||||
|
||||
// POST /api/etl/sync?mode=full|incremental
|
||||
router.post('/sync', async (req: Request, res: Response) => {
|
||||
// Auth check
|
||||
const auth = req.headers.authorization;
|
||||
if (etl.secret && auth !== `Bearer ${etl.secret}`) {
|
||||
res.status(401).json({ error: 'Unauthorized' });
|
||||
return;
|
||||
}
|
||||
|
||||
const mode = (req.query.mode as string) === 'full' ? 'full' : 'incremental';
|
||||
|
||||
try {
|
||||
console.log(`\nETL sync started (${mode})...`);
|
||||
const result = await runSync(mode);
|
||||
console.log(`ETL sync complete: ${result.transactionsFetched} transactions → ${result.recordsWritten} records in ${result.duration}`);
|
||||
res.json(result);
|
||||
} catch (err) {
|
||||
console.error('ETL sync failed:', (err as Error).message);
|
||||
res.status(500).json({ error: 'ETL sync failed', details: (err as Error).message });
|
||||
}
|
||||
});
|
||||
|
||||
// GET /api/etl/status
|
||||
router.get('/status', (_req: Request, res: Response) => {
|
||||
res.json({ configured: !!etl.secret });
|
||||
});
|
||||
|
||||
export default router;
|
||||
119
server/src/services/etlSync.ts
Normal file
119
server/src/services/etlSync.ts
Normal file
@@ -0,0 +1,119 @@
|
||||
import { fetchSales } from './erpClient';
|
||||
import { discoverTableIds, deleteRowsByMonth, deleteAllRows, insertRecords } from './nocodbClient';
|
||||
import { getMuseumFromProduct, getChannelLabel } from '../config/museumMapping';
|
||||
import type { ERPSaleRecord, AggregatedRecord } from '../types';
|
||||
|
||||
function generateMonthBoundaries(startYear: number, startMonth: number): Array<[string, string]> {
|
||||
const now = new Date();
|
||||
const endYear = now.getFullYear();
|
||||
const endMonth = now.getMonth() + 1;
|
||||
const boundaries: Array<[string, string]> = [];
|
||||
|
||||
let y = startYear;
|
||||
let m = startMonth;
|
||||
while (y < endYear || (y === endYear && m <= endMonth)) {
|
||||
const start = `${y}-${String(m).padStart(2, '0')}-01T00:00:00`;
|
||||
const nextM = m === 12 ? 1 : m + 1;
|
||||
const nextY = m === 12 ? y + 1 : y;
|
||||
const end = `${nextY}-${String(nextM).padStart(2, '0')}-01T00:00:00`;
|
||||
boundaries.push([start, end]);
|
||||
y = nextY;
|
||||
m = nextM;
|
||||
}
|
||||
|
||||
return boundaries;
|
||||
}
|
||||
|
||||
function currentMonthBoundary(): [string, string] {
|
||||
const now = new Date();
|
||||
const y = now.getFullYear();
|
||||
const m = now.getMonth() + 1;
|
||||
const start = `${y}-${String(m).padStart(2, '0')}-01T00:00:00`;
|
||||
const nextM = m === 12 ? 1 : m + 1;
|
||||
const nextY = m === 12 ? y + 1 : y;
|
||||
const end = `${nextY}-${String(nextM).padStart(2, '0')}-01T00:00:00`;
|
||||
return [start, end];
|
||||
}
|
||||
|
||||
export function aggregateTransactions(sales: ERPSaleRecord[]): AggregatedRecord[] {
|
||||
const map = new Map<string, AggregatedRecord>();
|
||||
|
||||
for (const sale of sales) {
|
||||
const date = sale.TransactionDate.split(' ')[0];
|
||||
const channel = getChannelLabel(sale.OperatingAreaName);
|
||||
|
||||
for (const product of sale.Products) {
|
||||
const museum = getMuseumFromProduct(product.ProductDescription);
|
||||
const key = `${date}|${museum}|${channel}`;
|
||||
|
||||
let entry = map.get(key);
|
||||
if (!entry) {
|
||||
entry = { Date: date, MuseumName: museum, Channel: channel, Visits: 0, Tickets: 0, GrossRevenue: 0, NetRevenue: 0 };
|
||||
map.set(key, entry);
|
||||
}
|
||||
|
||||
entry.Visits += product.PeopleCount;
|
||||
entry.Tickets += product.UnitQuantity;
|
||||
entry.GrossRevenue += product.TotalPrice;
|
||||
entry.NetRevenue += product.TotalPrice - product.TaxAmount;
|
||||
}
|
||||
}
|
||||
|
||||
return Array.from(map.values());
|
||||
}
|
||||
|
||||
export interface SyncResult {
|
||||
status: string;
|
||||
mode: string;
|
||||
transactionsFetched: number;
|
||||
recordsWritten: number;
|
||||
duration: string;
|
||||
}
|
||||
|
||||
export async function runSync(mode: 'full' | 'incremental' = 'incremental'): Promise<SyncResult> {
|
||||
const start = Date.now();
|
||||
|
||||
const tables = await discoverTableIds();
|
||||
const tableId = tables['DailySales'];
|
||||
if (!tableId) throw new Error("NocoDB table 'DailySales' not found");
|
||||
|
||||
let months: Array<[string, string]>;
|
||||
if (mode === 'full') {
|
||||
months = generateMonthBoundaries(2024, 1);
|
||||
} else {
|
||||
months = [currentMonthBoundary()];
|
||||
}
|
||||
|
||||
// Fetch from ERP sequentially (API can't handle concurrent requests)
|
||||
const allSales: ERPSaleRecord[] = [];
|
||||
for (const [startDate, endDate] of months) {
|
||||
console.log(` Fetching ${startDate.slice(0, 7)}...`);
|
||||
const chunk = await fetchSales(startDate, endDate) as ERPSaleRecord[];
|
||||
allSales.push(...chunk);
|
||||
}
|
||||
|
||||
const records = aggregateTransactions(allSales);
|
||||
|
||||
// Write to NocoDB
|
||||
if (mode === 'full') {
|
||||
console.log(' Clearing all DailySales rows...');
|
||||
await deleteAllRows(tableId);
|
||||
} else {
|
||||
const yearMonth = months[0][0].slice(0, 7);
|
||||
console.log(` Clearing ${yearMonth} rows...`);
|
||||
await deleteRowsByMonth(tableId, yearMonth);
|
||||
}
|
||||
|
||||
console.log(` Inserting ${records.length} records...`);
|
||||
const written = await insertRecords(tableId, records);
|
||||
|
||||
const duration = ((Date.now() - start) / 1000).toFixed(1) + 's';
|
||||
|
||||
return {
|
||||
status: 'ok',
|
||||
mode,
|
||||
transactionsFetched: allSales.length,
|
||||
recordsWritten: written,
|
||||
duration,
|
||||
};
|
||||
}
|
||||
109
server/src/services/nocodbClient.ts
Normal file
109
server/src/services/nocodbClient.ts
Normal file
@@ -0,0 +1,109 @@
|
||||
import { nocodb } from '../config';
|
||||
import type { AggregatedRecord } from '../types';
|
||||
|
||||
let discoveredTables: Record<string, string> | null = null;
|
||||
|
||||
async function fetchJson(url: string, options: RequestInit = {}): Promise<unknown> {
|
||||
const res = await fetch(url, {
|
||||
...options,
|
||||
headers: {
|
||||
'xc-token': nocodb.token,
|
||||
'Content-Type': 'application/json',
|
||||
...options.headers,
|
||||
},
|
||||
});
|
||||
if (!res.ok) {
|
||||
const text = await res.text().catch(() => '');
|
||||
throw new Error(`NocoDB ${res.status}: ${text.slice(0, 200)}`);
|
||||
}
|
||||
return res.json();
|
||||
}
|
||||
|
||||
export async function discoverTableIds(): Promise<Record<string, string>> {
|
||||
if (discoveredTables) return discoveredTables;
|
||||
if (!nocodb.baseId) throw new Error('NOCODB_BASE_ID not configured');
|
||||
|
||||
const json = await fetchJson(
|
||||
`${nocodb.url}/api/v2/meta/bases/${nocodb.baseId}/tables`
|
||||
) as { list: Array<{ title: string; id: string }> };
|
||||
|
||||
const tables: Record<string, string> = {};
|
||||
for (const t of json.list) {
|
||||
tables[t.title] = t.id;
|
||||
}
|
||||
|
||||
discoveredTables = tables;
|
||||
return tables;
|
||||
}
|
||||
|
||||
export async function deleteRowsByMonth(tableId: string, yearMonth: string): Promise<number> {
|
||||
// Fetch all row IDs for the given month using a where filter
|
||||
const where = `(Date,like,${yearMonth}%)`;
|
||||
let deleted = 0;
|
||||
let hasMore = true;
|
||||
|
||||
while (hasMore) {
|
||||
const json = await fetchJson(
|
||||
`${nocodb.url}/api/v2/tables/${tableId}/records?where=${encodeURIComponent(where)}&limit=100&fields=Id`
|
||||
) as { list: Array<{ Id: number }> };
|
||||
|
||||
const ids = json.list.map(r => r.Id);
|
||||
if (ids.length === 0) {
|
||||
hasMore = false;
|
||||
break;
|
||||
}
|
||||
|
||||
// Bulk delete
|
||||
await fetchJson(`${nocodb.url}/api/v2/tables/${tableId}/records`, {
|
||||
method: 'DELETE',
|
||||
body: JSON.stringify(ids.map(Id => ({ Id }))),
|
||||
});
|
||||
|
||||
deleted += ids.length;
|
||||
}
|
||||
|
||||
return deleted;
|
||||
}
|
||||
|
||||
export async function deleteAllRows(tableId: string): Promise<number> {
|
||||
let deleted = 0;
|
||||
let hasMore = true;
|
||||
|
||||
while (hasMore) {
|
||||
const json = await fetchJson(
|
||||
`${nocodb.url}/api/v2/tables/${tableId}/records?limit=100&fields=Id`
|
||||
) as { list: Array<{ Id: number }> };
|
||||
|
||||
const ids = json.list.map(r => r.Id);
|
||||
if (ids.length === 0) {
|
||||
hasMore = false;
|
||||
break;
|
||||
}
|
||||
|
||||
await fetchJson(`${nocodb.url}/api/v2/tables/${tableId}/records`, {
|
||||
method: 'DELETE',
|
||||
body: JSON.stringify(ids.map(Id => ({ Id }))),
|
||||
});
|
||||
|
||||
deleted += ids.length;
|
||||
}
|
||||
|
||||
return deleted;
|
||||
}
|
||||
|
||||
export async function insertRecords(tableId: string, records: AggregatedRecord[]): Promise<number> {
|
||||
// NocoDB bulk insert accepts max 100 records at a time
|
||||
const batchSize = 100;
|
||||
let inserted = 0;
|
||||
|
||||
for (let i = 0; i < records.length; i += batchSize) {
|
||||
const batch = records.slice(i, i + batchSize);
|
||||
await fetchJson(`${nocodb.url}/api/v2/tables/${tableId}/records`, {
|
||||
method: 'POST',
|
||||
body: JSON.stringify(batch),
|
||||
});
|
||||
inserted += batch.length;
|
||||
}
|
||||
|
||||
return inserted;
|
||||
}
|
||||
27
server/src/types.ts
Normal file
27
server/src/types.ts
Normal file
@@ -0,0 +1,27 @@
|
||||
export interface ERPProduct {
|
||||
ProductDescription: string;
|
||||
SiteDescription: string | null;
|
||||
UnitQuantity: number;
|
||||
PeopleCount: number;
|
||||
TaxAmount: number;
|
||||
TotalPrice: number;
|
||||
}
|
||||
|
||||
export interface ERPSaleRecord {
|
||||
SaleId: number;
|
||||
TransactionDate: string;
|
||||
CustIdentification: string;
|
||||
OperatingAreaName: string;
|
||||
Payments: Array<{ PaymentMethodDescription: string }>;
|
||||
Products: ERPProduct[];
|
||||
}
|
||||
|
||||
export interface AggregatedRecord {
|
||||
Date: string;
|
||||
MuseumName: string;
|
||||
Channel: string;
|
||||
Visits: number;
|
||||
Tickets: number;
|
||||
GrossRevenue: number;
|
||||
NetRevenue: number;
|
||||
}
|
||||
Reference in New Issue
Block a user