jetour-bi/n8n/wf4_otps_updated.ts

import { workflow, node, trigger, expr } from '@n8n/workflow-sdk';

const cron = trigger({ type: 'n8n-nodes-base.scheduleTrigger', version: 1.3,
  config: { name: 'Daily 02:50', parameters: { rule: { interval: [{ field: 'cronExpression', expression: '50 2 * * *' }] } } } });

const auth = node({ type: 'n8n-nodes-base.httpRequest', version: 4.4,
  config: { name: 'Auth CMS BI', parameters: { method: 'POST', url: 'https://aws-prod-auth.cms360.cloud/api/Jwt/FromBody',
    sendBody: true, bodyParameters: { parameters: [{ name: 'username', value: 'JetourMauritiusBI' }, { name: 'password', value: 'j7PW0ev66mJ2' }] }, options: { timeout: 30000 } } } });

const setVars = node({ type: 'n8n-nodes-base.set', version: 3.4,
  config: { name: 'Set Dates + Token', parameters: { mode: 'manual', includeOtherFields: false,
    assignments: { assignments: [
      { id: 't', name: 'token', value: expr('={{ $json.token }}'), type: 'string' },
      { id: 'f', name: 'from', value: expr('={{ DateTime.now().minus({ days: 1 }).startOf("day").toISO() }}'), type: 'string' },
      { id: 't2', name: 'to', value: expr('={{ DateTime.now().startOf("day").toISO() }}'), type: 'string' }
    ] } } } });

const apiCall = node({ type: 'n8n-nodes-base.httpRequest', version: 4.4,
  config: { name: 'Fetch OTPs', parameters: { method: 'POST',
    url: 'https://bi-api.cmscloud.co.za/api/BusinessIntelligence/Otps/GetOtpsByDateUpdated',
    sendBody: true, sendHeaders: true, specifyHeaders: 'keypair',
    headerParameters: { parameters: [
      { name: 'Authorization', value: expr('=Bearer {{ $("Set Dates + Token").item.json.token }}') },
      { name: 'Content-Type', value: 'application/json' }] },
    bodyParameters: { parameters: [
      { id: 'f', name: 'from', value: expr('={{ $("Set Dates + Token").item.json.from }}') },
      { id: 't2', name: 'to', value: expr('={{ $("Set Dates + Token").item.json.to }}') },
      { name: 'pageNumber', value: '1' }, { name: 'pageSize', value: '500' }] },
    options: { timeout: 120000 } } } });

const flatten = node({ type: 'n8n-nodes-base.code', version: 2,
  config: { name: 'Flatten', parameters: { mode: 'runOnceForAllItems', jsCode:
`const items = $input.all();
const ts = new Date().toISOString().replace(/[-:T]/g,'').slice(0,15);
const bid = "ot_upd_" + ts;
const out = [];
function n(v) { return v || null; }
function pu(v) { if(!v) return null; return v.replace(/[{}]/g,'')||null; }
for (const it of items) {
  const d = it.json;
  out.push({ json: { _t:'P', pid:d.prospectID, cid:d.contactID, cdr:n(d.cmsDealerRef), kci:n(d.keyloopCustID),
    kca:n(d.keyloopCustAccNumber), dd:n(d.dealerDesc), cn:n(d.companyName), ti:n(d.title),
    nm:n(d.name), sn:n(d.surname), db:n(d.dtBirth), lg:n(d.language), cl:n(d.cellNumber),
    em:n(d.emailAddr), sc:n(d.sourceOfInitialContact), rs:n(d.referralSource),
    sp:n(d.salesPerson), ps:n(d.prospectStatus), lr:n(d.lostReason), bim:n(d.basicInterestMake),
    bimd:n(d.basicInterestModel), nu:n(d.newUsed), dc:n(d.dtProspectCreated),
    du:n(d.dtProspectUpdated), da:n(d.dtLastAction), bid:bid },
    pairedItem: { item: it.index } });
  if (d.prospectOTPS) for (const o of d.prospectOTPS) {
    out.push({ json: { _t:'O', pid:d.prospectID, oid:o.cmsotpid, cpid:o.cmsProspectID,
      eg:pu(o.exportGUID), on:n(o.otpNewUsed), mc:n(o.mmCode), mk:n(o.make), md:n(o.model),
      mr:n(o.modelRange), co:n(o.colour), my:n(o.modelYear), vi:n(o.vin), sn:n(o.stockNo),
      am:o.amount, dc:n(o.dtotpCreated), ds:n(o.dtSold), di:n(o.dtInvoiced), bid:bid },
      pairedItem: { item: it.index } });
  }
  if (d.statusLog) for (const s of d.statusLog) {
    out.push({ json: { _t:'S', pid:d.prospectID, st:s.status, dt:n(s.dtUpdated), bid:bid },
    pairedItem: { item: it.index } });
  }
}
return out;` } } });

const insertP = node({ type: 'n8n-nodes-base.postgres', version: 2.4,
  config: { name: 'Upsert Prospects', parameters: { operation: 'executeQuery', query:
    expr(`=INSERT INTO prospects (prospect_id,contact_id,cms_dealer_ref,keyloop_cust_id,keyloop_cust_acc_number,dealer_desc,company_name,title,name,surname,dt_birth,language,cell_number,email_addr,source_of_initial_contact,referral_source,sales_person,prospect_status,lost_reason,basic_interest_make,basic_interest_model,new_used,dt_prospect_created,dt_prospect_updated,dt_last_action,import_batch_id)
VALUES ({{ $json.pid }},{{ $json.cid || 'NULL' }},{{ $json.cdr ? "'"+$json.cdr.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.kci ? "'"+$json.kci+"'" : 'NULL' }},{{ $json.kca ? "'"+$json.kca+"'" : 'NULL' }},{{ $json.dd ? "'"+$json.dd.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.cn ? "'"+$json.cn.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.ti ? "'"+$json.ti.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.nm ? "'"+$json.nm.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.sn ? "'"+$json.sn.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.db ? "'"+$json.db+"'::date" : 'NULL' }},{{ $json.lg ? "'"+$json.lg+"'" : 'NULL' }},{{ $json.cl ? "'"+$json.cl+"'" : 'NULL' }},{{ $json.em ? "'"+$json.em+"'" : 'NULL' }},{{ $json.sc ? "'"+$json.sc.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.rs ? "'"+$json.rs.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.sp ? "'"+$json.sp.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.ps ? "'"+$json.ps.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.lr ? "'"+$json.lr.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.bim ? "'"+$json.bim.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.bimd ? "'"+$json.bimd.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.nu ? "'"+$json.nu+"'" : 'NULL' }},{{ $json.dc ? "'"+$json.dc+"'::timestamptz" : 'NULL' }},{{ $json.du ? "'"+$json.du+"'::timestamptz" : 'NULL' }},{{ $json.da ? "'"+$json.da+"'::timestamptz" : 'NULL' }},{{ "'"+$json.bid+"'" }})
ON CONFLICT (prospect_id) DO NOTHING;`) },
  credentials: { postgres: { name: 'jetour-bi-remote' } } } });

const insertO = node({ type: 'n8n-nodes-base.postgres', version: 2.4,
  config: { name: 'Upsert OTPs', parameters: { operation: 'executeQuery', query:
    expr(`=INSERT INTO prospect_otps (prospect_id,cms_otp_id,cms_prospect_id,export_guid,otp_new_used,mm_code,make,model,model_range,colour,model_year,vin,stock_no,amount,dt_otp_created,dt_sold,dt_invoiced,import_batch_id)
VALUES ((SELECT id FROM prospects WHERE prospect_id={{ $json.pid }}),{{ $json.oid }},{{ $json.cpid }},{{ $json.eg ? "'"+$json.eg+"'::uuid" : 'NULL' }},{{ $json.on ? "'"+$json.on+"'" : 'NULL' }},{{ $json.mc ? "'"+$json.mc+"'" : 'NULL' }},{{ $json.mk ? "'"+$json.mk.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.md ? "'"+$json.md.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.mr ? "'"+$json.mr.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.co ? "'"+$json.co.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.my ? "'"+$json.my+"'" : 'NULL' }},{{ $json.vi ? "'"+$json.vi+"'" : 'NULL' }},{{ $json.sn ? "'"+$json.sn.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.am !== null && $json.am !== undefined ? $json.am : 'NULL' }},{{ $json.dc ? "'"+$json.dc+"'::timestamptz" : 'NULL' }},{{ $json.ds ? "'"+$json.ds+"'::timestamptz" : 'NULL' }},{{ $json.di ? "'"+$json.di+"'::timestamptz" : 'NULL' }},{{ "'"+$json.bid+"'" }})
ON CONFLICT (cms_otp_id) DO UPDATE SET prospect_id=EXCLUDED.prospect_id, cms_prospect_id=EXCLUDED.cms_prospect_id, export_guid=EXCLUDED.export_guid, otp_new_used=EXCLUDED.otp_new_used, mm_code=EXCLUDED.mm_code, make=EXCLUDED.make, model=EXCLUDED.model, model_range=EXCLUDED.model_range, colour=EXCLUDED.colour, model_year=EXCLUDED.model_year, vin=EXCLUDED.vin, stock_no=EXCLUDED.stock_no, amount=EXCLUDED.amount, dt_sold=EXCLUDED.dt_sold, dt_invoiced=EXCLUDED.dt_invoiced, last_imported_at=NOW(), import_batch_id=EXCLUDED.import_batch_id;`) },
  credentials: { postgres: { name: 'jetour-bi-remote' } } } });

const insertS = node({ type: 'n8n-nodes-base.postgres', version: 2.4,
  config: { name: 'Upsert Status Logs', parameters: { operation: 'executeQuery', query:
    expr(`=INSERT INTO status_logs (prospect_id,status,dt_updated,import_batch_id)
VALUES ((SELECT id FROM prospects WHERE prospect_id={{ $json.pid }}),{{ $json.st ? "'"+$json.st.replace(/'/g,"''")+"'" : 'NULL' }},{{ $json.dt ? "'"+$json.dt+"'::timestamptz" : 'NULL' }},{{ "'"+$json.bid+"'" }})
ON CONFLICT (prospect_id,status,dt_updated) DO NOTHING;`) },
  credentials: { postgres: { name: 'jetour-bi-remote' } } } });

export default workflow('CMS_BI_OTPS_UPDATED', 'CMS BI: GetOtpsByDateUpdated → jetour-bi')
  .add(cron).to(auth).to(setVars).to(apiCall).to(flatten)
  .add(flatten).to(insertP).add(flatten).to(insertO).add(flatten).to(insertS);