Uses /api/v2/timesheets/on/{date} per day (no location param), expands
nested shifts[], skips leave (leave_request_id!=null), and post-filters
to this hotel's departments via location_id match on the departments
list. This mirrors exactly what the FD's Power BI query does and should
give the correct £51k timesheet figure for June.
Also simplifies compareEndpoints to two variants: A (old shifts method)
vs B (new timesheets method) so the discrepancy is immediately visible.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
441 lines
17 KiB
JavaScript
441 lines
17 KiB
JavaScript
import { pool, getConfig } from '../db.js'
|
|
|
|
const SETTINGS_URL = process.env.SETTINGS_URL || 'http://10.10.10.116:3080'
|
|
const SETTINGS_SECRET = process.env.SETTINGS_SECRET || ''
|
|
|
|
let _credsCache = null
|
|
|
|
async function getWorkforceCreds() {
|
|
if (_credsCache && Date.now() < _credsCache.expires_at) return _credsCache.creds
|
|
const res = await fetch(`${SETTINGS_URL}/settings/api/internal/integration/workforce`, {
|
|
headers: { Authorization: `Bearer ${SETTINGS_SECRET}` },
|
|
signal: AbortSignal.timeout(5000),
|
|
})
|
|
if (!res.ok) throw new Error('Workforce integration not configured — add bearer token in Settings')
|
|
const creds = await res.json()
|
|
if (!creds.bearer_token) throw new Error('Workforce integration not configured — add bearer token in Settings')
|
|
_credsCache = { creds, expires_at: Date.now() + 5 * 60_000 }
|
|
return creds
|
|
}
|
|
|
|
async function wfFetch(path) {
|
|
const creds = await getWorkforceCreds()
|
|
const base = creds.base_url || 'https://my.workforce.com'
|
|
const res = await fetch(`${base}${path}`, {
|
|
headers: { Authorization: `Bearer ${creds.bearer_token}` },
|
|
signal: AbortSignal.timeout(15000),
|
|
})
|
|
if (!res.ok) {
|
|
const body = await res.text().catch(() => '')
|
|
throw new Error(`Workforce API ${res.status}${body ? ': ' + body.slice(0, 200) : ''}`)
|
|
}
|
|
return res.json()
|
|
}
|
|
|
|
async function wfFetchPaged(path) {
|
|
const results = []
|
|
let page = 1
|
|
while (true) {
|
|
const sep = path.includes('?') ? '&' : '?'
|
|
const data = await wfFetch(`${path}${sep}page=${page}&page_size=100`)
|
|
const items = Array.isArray(data)
|
|
? data
|
|
: (data.users ?? data.departments ?? data.schedules ?? data.shifts ?? data.timesheets ?? [])
|
|
results.push(...items)
|
|
if (items.length < 100) break
|
|
page++
|
|
}
|
|
return results
|
|
}
|
|
|
|
export async function fetchAllDepartments() {
|
|
const creds = await getWorkforceCreds()
|
|
const locationId = creds.location_id ? String(creds.location_id) : null
|
|
const all = await wfFetchPaged('/api/v2/departments')
|
|
const filtered = locationId ? all.filter(d => String(d.location_id) === locationId) : all
|
|
return filtered.map(d => ({ id: String(d.id), name: d.name }))
|
|
}
|
|
|
|
async function getEnabledDeptIds() {
|
|
const val = await getConfig('departments')
|
|
if (!val) return null // null means "all enabled"
|
|
try {
|
|
const depts = JSON.parse(val)
|
|
if (!Array.isArray(depts) || depts.length === 0) return null
|
|
const enabled = depts.filter(d => d.enabled !== false).map(d => d.id)
|
|
return enabled.length > 0 ? enabled : null
|
|
} catch {
|
|
return null
|
|
}
|
|
}
|
|
|
|
async function getUserNameMap() {
|
|
const all = await wfFetchPaged('/api/v2/users')
|
|
return Object.fromEntries(
|
|
all.map(u => [String(u.id), u.name || `${u.legal_first_name || ''} ${u.legal_last_name || ''}`.trim() || `User ${u.id}`])
|
|
)
|
|
}
|
|
|
|
// Build list of YYYY-MM-DD strings between from and to (inclusive)
|
|
function dateRange(from, to) {
|
|
const dates = []
|
|
const cur = new Date(from + 'T00:00:00')
|
|
const end = new Date(to + 'T00:00:00')
|
|
while (cur <= end) { dates.push(cur.toISOString().slice(0, 10)); cur.setDate(cur.getDate() + 1) }
|
|
return dates
|
|
}
|
|
|
|
// Fetch timesheets for a single date — no location param (mirrors PBI pattern).
|
|
// Each record is { user_id, shifts: [{ department_id, date, cost, leave_request_id, ... }] }
|
|
async function fetchTimesheetsForDate(dateStr) {
|
|
try { return await wfFetchPaged(`/api/v2/timesheets/on/${dateStr}?show_costs=true`) }
|
|
catch { return [] }
|
|
}
|
|
|
|
// Expand an array of timesheet objects into flat shift rows, filtering out leave and
|
|
// optionally filtering by location department IDs and enabled department IDs.
|
|
function expandShifts(timesheets, locationDeptIds, enabledDeptIds) {
|
|
const rows = []
|
|
for (const t of timesheets) {
|
|
if (!Array.isArray(t.shifts)) continue
|
|
const userId = String(t.user_id)
|
|
for (const sh of t.shifts) {
|
|
if (sh.leave_request_id != null) continue // exclude leave accrual shifts
|
|
const deptId = String(sh.department_id ?? 'unknown')
|
|
if (locationDeptIds && !locationDeptIds.has(deptId)) continue // filter to this hotel
|
|
if (enabledDeptIds && !enabledDeptIds.includes(deptId)) continue
|
|
rows.push({ date: sh.date, userId, deptId, cost: parseFloat(sh.cost ?? 0) })
|
|
}
|
|
}
|
|
return rows
|
|
}
|
|
|
|
export async function syncActuals(from, to) {
|
|
const creds = await getWorkforceCreds()
|
|
const locationId = creds.location_id ? String(creds.location_id) : null
|
|
const enabledDeptIds = await getEnabledDeptIds()
|
|
|
|
// Fetch departments + users in parallel — departments used to filter to this hotel's depts
|
|
const [allDepts, userNameMap] = await Promise.all([
|
|
wfFetchPaged('/api/v2/departments'),
|
|
getUserNameMap(),
|
|
])
|
|
const deptNameMap = Object.fromEntries(allDepts.map(d => [String(d.id), d.name]))
|
|
const locationDeptIds = locationId
|
|
? new Set(allDepts.filter(d => String(d.location_id) === locationId).map(d => String(d.id)))
|
|
: null
|
|
|
|
// Fetch all dates in parallel — same pattern as PBI (no location param on API)
|
|
const dates = dateRange(from, to)
|
|
const perDay = await Promise.all(dates.map(fetchTimesheetsForDate))
|
|
|
|
const byDateDept = {}
|
|
const byDateDeptEmp = {}
|
|
|
|
for (const timesheets of perDay) {
|
|
for (const { date, userId, deptId, cost } of expandShifts(timesheets, locationDeptIds, enabledDeptIds)) {
|
|
const key = `${date}:${deptId}`
|
|
if (!byDateDept[key]) {
|
|
byDateDept[key] = {
|
|
date, department_id: deptId,
|
|
department_name: deptNameMap[deptId] || deptId,
|
|
base_cost: 0, total_cost: 0, shift_count: 0,
|
|
}
|
|
}
|
|
byDateDept[key].base_cost += cost
|
|
byDateDept[key].total_cost += cost
|
|
byDateDept[key].shift_count += 1
|
|
|
|
if (cost > 0) {
|
|
const empKey = `${date}:${deptId}:${userId}`
|
|
const empName = userNameMap[userId] || `Employee ${userId}`
|
|
if (!byDateDeptEmp[empKey]) {
|
|
byDateDeptEmp[empKey] = {
|
|
date, department_id: deptId, employee_id: userId, employee_name: empName,
|
|
base_cost: 0, total_cost: 0, shift_count: 0,
|
|
}
|
|
}
|
|
byDateDeptEmp[empKey].base_cost += cost
|
|
byDateDeptEmp[empKey].total_cost += cost
|
|
byDateDeptEmp[empKey].shift_count += 1
|
|
}
|
|
}
|
|
}
|
|
|
|
for (const row of Object.values(byDateDept)) {
|
|
await pool.query(
|
|
`INSERT INTO wage_actuals (date, department_id, department_name, base_cost, total_cost, shift_count, cached_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, NOW())
|
|
ON CONFLICT (date, department_id) DO UPDATE SET
|
|
department_name = EXCLUDED.department_name,
|
|
base_cost = EXCLUDED.base_cost,
|
|
total_cost = EXCLUDED.total_cost,
|
|
shift_count = EXCLUDED.shift_count,
|
|
cached_at = NOW()`,
|
|
[row.date, row.department_id, row.department_name,
|
|
row.base_cost.toFixed(2), row.total_cost.toFixed(2), row.shift_count]
|
|
)
|
|
}
|
|
|
|
for (const row of Object.values(byDateDeptEmp)) {
|
|
await pool.query(
|
|
`INSERT INTO wage_actuals_detail (date, department_id, employee_id, employee_name, base_cost, total_cost, shift_count, cached_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())
|
|
ON CONFLICT (date, department_id, employee_id) DO UPDATE SET
|
|
employee_name = EXCLUDED.employee_name,
|
|
base_cost = EXCLUDED.base_cost,
|
|
total_cost = EXCLUDED.total_cost,
|
|
shift_count = EXCLUDED.shift_count,
|
|
cached_at = NOW()`,
|
|
[row.date, row.department_id, row.employee_id, row.employee_name,
|
|
row.base_cost.toFixed(2), row.total_cost.toFixed(2), row.shift_count]
|
|
)
|
|
}
|
|
|
|
return Object.keys(byDateDept).length
|
|
}
|
|
|
|
export async function syncScheduled(from, to) {
|
|
const creds = await getWorkforceCreds()
|
|
const locationId = creds.location_id
|
|
const enabledDeptIds = await getEnabledDeptIds()
|
|
|
|
let path = `/api/v2/schedules?from=${from}&to=${to}&show_costs=true&include_oncosts=true`
|
|
if (locationId) path += `&location_id=${locationId}`
|
|
|
|
const schedules = await wfFetchPaged(path)
|
|
const deptNameMap = await getDeptNameMap()
|
|
|
|
const byDateDept = {}
|
|
for (const s of schedules) {
|
|
const deptId = String(s.department_id)
|
|
if (enabledDeptIds && !enabledDeptIds.includes(deptId)) continue
|
|
|
|
const date = s.date || new Date(s.start * 1000).toISOString().slice(0, 10)
|
|
const key = `${date}:${deptId}`
|
|
if (!byDateDept[key]) {
|
|
byDateDept[key] = {
|
|
date,
|
|
department_id: deptId,
|
|
department_name: deptNameMap[deptId] || deptId,
|
|
base_cost: 0,
|
|
total_cost: 0,
|
|
shift_count: 0,
|
|
}
|
|
}
|
|
byDateDept[key].base_cost += parseFloat(s.cost ?? 0)
|
|
byDateDept[key].total_cost += parseFloat(s.cost_with_oncosts ?? s.cost ?? 0)
|
|
byDateDept[key].shift_count += 1
|
|
}
|
|
|
|
for (const row of Object.values(byDateDept)) {
|
|
await pool.query(
|
|
`INSERT INTO wage_scheduled (date, department_id, department_name, base_cost, total_cost, shift_count, cached_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, NOW())
|
|
ON CONFLICT (date, department_id) DO UPDATE SET
|
|
department_name = EXCLUDED.department_name,
|
|
base_cost = EXCLUDED.base_cost,
|
|
total_cost = EXCLUDED.total_cost,
|
|
shift_count = EXCLUDED.shift_count,
|
|
cached_at = NOW()`,
|
|
[row.date, row.department_id, row.department_name,
|
|
row.base_cost.toFixed(2), row.total_cost.toFixed(2), row.shift_count]
|
|
)
|
|
}
|
|
|
|
return Object.keys(byDateDept).length
|
|
}
|
|
|
|
export async function runRollingSync() {
|
|
const today = new Date()
|
|
const todayStr = today.toISOString().slice(0, 10)
|
|
|
|
// Actuals: 35 days back in 7-day batches (API limit is 31 days per request)
|
|
const startDate = new Date(today)
|
|
startDate.setDate(startDate.getDate() - 35)
|
|
let actualRows = 0
|
|
let current = new Date(startDate)
|
|
while (current <= today) {
|
|
const batchEnd = new Date(current)
|
|
batchEnd.setDate(batchEnd.getDate() + 6)
|
|
if (batchEnd > today) batchEnd.setTime(today.getTime())
|
|
actualRows += await syncActuals(
|
|
current.toISOString().slice(0, 10),
|
|
batchEnd.toISOString().slice(0, 10)
|
|
)
|
|
current.setDate(current.getDate() + 7)
|
|
}
|
|
|
|
// Scheduled: next 14 days in 7-day batches
|
|
const schedEnd = new Date(today)
|
|
schedEnd.setDate(schedEnd.getDate() + 14)
|
|
let scheduledRows = 0
|
|
let schedCurrent = new Date(today)
|
|
while (schedCurrent <= schedEnd) {
|
|
const batchEnd = new Date(schedCurrent)
|
|
batchEnd.setDate(batchEnd.getDate() + 6)
|
|
if (batchEnd > schedEnd) batchEnd.setTime(schedEnd.getTime())
|
|
scheduledRows += await syncScheduled(
|
|
schedCurrent.toISOString().slice(0, 10),
|
|
batchEnd.toISOString().slice(0, 10)
|
|
)
|
|
schedCurrent.setDate(schedCurrent.getDate() + 7)
|
|
}
|
|
|
|
return { actualRows, scheduledRows }
|
|
}
|
|
|
|
// Fetch a raw shifts list from a path and return a cost summary — used for endpoint comparison
|
|
async function fetchShiftsAndSummarise(path, locationDeptIds) {
|
|
try {
|
|
const shifts = await wfFetchPaged(path)
|
|
if (!Array.isArray(shifts) || shifts.length === 0) return { count: 0, baseCost: 0, topDepts: [] }
|
|
let baseCost = 0
|
|
const byDept = {}
|
|
let sampleShift = null
|
|
for (const s of shifts) {
|
|
const deptId = String(s.department_id ?? 'unknown')
|
|
if (locationDeptIds && !locationDeptIds.has(deptId)) continue
|
|
const cost = parseFloat(s.cost ?? 0)
|
|
baseCost += cost
|
|
if (!sampleShift && cost > 0) sampleShift = s
|
|
if (!byDept[deptId]) byDept[deptId] = { name: s.department_name || deptId, cost: 0, count: 0 }
|
|
byDept[deptId].cost += cost
|
|
byDept[deptId].count++
|
|
}
|
|
const topDepts = Object.entries(byDept)
|
|
.map(([id, v]) => ({ id, name: v.name, cost: +v.cost.toFixed(2), count: v.count }))
|
|
.sort((a, b) => b.cost - a.cost)
|
|
const result = { rawCount: shifts.length, filteredCount: Object.values(byDept).reduce((s, v) => s + v.count, 0), baseCost: +baseCost.toFixed(2), topDepts: topDepts.slice(0, 15) }
|
|
if (sampleShift) result.sampleShiftFields = Object.fromEntries(Object.entries(sampleShift).filter(([, v]) => v !== null && v !== undefined && v !== ''))
|
|
return result
|
|
} catch (err) {
|
|
return { error: err.message }
|
|
}
|
|
}
|
|
|
|
// Fetch per-date timesheets for every day in [from, to] in parallel.
|
|
// Expands nested shifts[], filters leave, optionally filters by location dept IDs.
|
|
// This is the method used by PBI — no location param on the API call.
|
|
async function fetchTimesheetsByDay(from, to, locationDeptIds) {
|
|
const dates = dateRange(from, to)
|
|
const perDay = await Promise.all(dates.map(fetchTimesheetsForDate))
|
|
|
|
let totalShifts = 0, filteredShifts = 0, cost = 0
|
|
const byDept = {}
|
|
let sampleShift = null
|
|
|
|
for (const timesheets of perDay) {
|
|
if (!Array.isArray(timesheets)) continue
|
|
for (const t of timesheets) {
|
|
if (!Array.isArray(t.shifts)) continue
|
|
for (const sh of t.shifts) {
|
|
totalShifts++
|
|
const isLeave = sh.leave_request_id != null
|
|
const deptId = String(sh.department_id ?? 'unknown')
|
|
const inLoc = !locationDeptIds || locationDeptIds.has(deptId)
|
|
if (isLeave || !inLoc) continue
|
|
filteredShifts++
|
|
const shCost = parseFloat(sh.cost ?? 0)
|
|
cost += shCost
|
|
if (!sampleShift && shCost > 0) sampleShift = sh
|
|
if (!byDept[deptId]) byDept[deptId] = { cost: 0, count: 0 }
|
|
byDept[deptId].cost += shCost
|
|
byDept[deptId].count++
|
|
}
|
|
}
|
|
}
|
|
|
|
const topDepts = Object.entries(byDept)
|
|
.map(([id, v]) => ({ id, cost: +v.cost.toFixed(2), count: v.count }))
|
|
.sort((a, b) => b.cost - a.cost)
|
|
|
|
return {
|
|
timesheetRecords: perDay.reduce((s, d) => s + (Array.isArray(d) ? d.length : 0), 0),
|
|
totalShifts,
|
|
filteredShifts,
|
|
cost: +cost.toFixed(2),
|
|
topDepts: topDepts.slice(0, 15),
|
|
sampleShiftFields: sampleShift
|
|
? Object.fromEntries(Object.entries(sampleShift).filter(([, v]) => v !== null && v !== undefined && v !== ''))
|
|
: null,
|
|
}
|
|
}
|
|
|
|
// Compare cost totals from available endpoints for a given date range.
|
|
// GET /api/sync/compare?from=YYYY-MM-DD&to=YYYY-MM-DD
|
|
export async function compareEndpoints(from, to) {
|
|
const creds = await getWorkforceCreds()
|
|
const locationId = creds.location_id ? String(creds.location_id) : null
|
|
|
|
// Build location dept IDs set for post-fetch filtering (PBI pattern)
|
|
const allDepts = await wfFetchPaged('/api/v2/departments')
|
|
const locationDeptIds = locationId
|
|
? new Set(allDepts.filter(d => String(d.location_id) === locationId).map(d => String(d.id)))
|
|
: null
|
|
|
|
// A: Shifts with report_location_id (old sync method)
|
|
const shiftsReportLoc = `/api/v2/shifts?from=${from}&to=${to}&show_costs=true&include_oncosts=true`
|
|
+ (locationId ? `&report_location_id=${locationId}` : '')
|
|
|
|
// B: Timesheets per-day, no API location filter, post-filtered to this hotel's depts
|
|
// This is the PBI method and should match WF "Timesheet exc. leave inc. allowances"
|
|
const [a, b] = await Promise.all([
|
|
fetchShiftsAndSummarise(shiftsReportLoc, locationDeptIds),
|
|
fetchTimesheetsByDay(from, to, locationDeptIds),
|
|
])
|
|
|
|
return {
|
|
from, to,
|
|
location_id: locationId,
|
|
location_dept_count: locationDeptIds ? locationDeptIds.size : 'all',
|
|
A_shifts_report_location_id: {
|
|
...a,
|
|
note: 'Old sync method — shifts endpoint with report_location_id filter',
|
|
},
|
|
B_timesheets_per_day_no_leave: {
|
|
...b,
|
|
note: 'PBI method — timesheets/on/{date}, shifts expanded, leave excluded (leave_request_id!=null), filtered to this hotel\'s dept IDs. Target: WF Timesheet exc.leave inc.allowances',
|
|
},
|
|
wf_report_reference: {
|
|
note: 'Compare B.cost against WF "Timesheet exc. leave inc. allowances" figure',
|
|
scheduled_exc_leave: null,
|
|
timesheet_exc_leave_inc_all: null,
|
|
leave_accrual: null,
|
|
},
|
|
}
|
|
}
|
|
|
|
export async function runBackfill(onProgress, signal) {
|
|
const today = new Date()
|
|
const endDate = new Date(today)
|
|
endDate.setDate(endDate.getDate() - 1)
|
|
|
|
const startDate = new Date(today)
|
|
startDate.setMonth(startDate.getMonth() - 25)
|
|
|
|
const totalDays = Math.max(1, Math.ceil((endDate - startDate) / 86_400_000))
|
|
let processedDays = 0
|
|
|
|
let current = new Date(startDate)
|
|
while (current <= endDate) {
|
|
if (signal?.aborted) break
|
|
|
|
const weekEnd = new Date(current)
|
|
weekEnd.setDate(weekEnd.getDate() + 6)
|
|
if (weekEnd > endDate) weekEnd.setTime(endDate.getTime())
|
|
|
|
const from = current.toISOString().slice(0, 10)
|
|
const to = weekEnd.toISOString().slice(0, 10)
|
|
|
|
await syncActuals(from, to)
|
|
|
|
const daysInBatch = Math.ceil((weekEnd - current) / 86_400_000) + 1
|
|
processedDays += daysInBatch
|
|
|
|
onProgress?.({ processed: processedDays, total: totalDays, current: from })
|
|
|
|
current.setDate(current.getDate() + 7)
|
|
await new Promise(r => setTimeout(r, 250))
|
|
}
|
|
}
|