Import and calculate tweaks and fixes
This commit is contained in:
426
inventory-server/old/backfill-snapshots.js
Normal file
426
inventory-server/old/backfill-snapshots.js
Normal file
@@ -0,0 +1,426 @@
|
||||
const path = require('path');
|
||||
const fs = require('fs');
|
||||
const progress = require('../scripts/metrics-new/utils/progress'); // Assuming progress utils are here
|
||||
const { getConnection, closePool } = require('../scripts/metrics-new/utils/db'); // Assuming db utils are here
|
||||
const os = require('os'); // For detecting number of CPU cores
|
||||
|
||||
// --- Configuration ---
|
||||
const BATCH_SIZE_DAYS = 1; // Process 1 day per database function call
|
||||
const SQL_FUNCTION_FILE = path.resolve(__dirname, 'backfill_historical_snapshots.sql'); // Correct path
|
||||
const LOG_PROGRESS_INTERVAL_MS = 5000; // Update console progress roughly every 5 seconds
|
||||
const HISTORY_TYPE = 'backfill_snapshots'; // Identifier for history table
|
||||
const MAX_WORKERS = Math.max(1, Math.floor(os.cpus().length / 2)); // Use half of available CPU cores
|
||||
const USE_PARALLEL = false; // Set to true to enable parallel processing
|
||||
const PG_STATEMENT_TIMEOUT_MS = 1800000; // 30 minutes max per query
|
||||
|
||||
// --- Cancellation Handling ---
|
||||
let isCancelled = false;
|
||||
let runningQueryPromise = null; // To potentially track the active query
|
||||
|
||||
function requestCancellation() {
|
||||
if (!isCancelled) {
|
||||
isCancelled = true;
|
||||
console.warn('\nCancellation requested. Finishing current batch then stopping...');
|
||||
// Note: We are NOT forcefully cancelling the backend query anymore.
|
||||
}
|
||||
}
|
||||
|
||||
process.on('SIGINT', requestCancellation); // Handle Ctrl+C
|
||||
process.on('SIGTERM', requestCancellation); // Handle termination signals
|
||||
|
||||
// --- Main Backfill Function ---
|
||||
async function backfillSnapshots(cmdStartDate, cmdEndDate, cmdStartBatch = 1) {
|
||||
let connection;
|
||||
const overallStartTime = Date.now();
|
||||
let calculateHistoryId = null;
|
||||
let processedDaysTotal = 0; // Track total days processed across all batches executed in this run
|
||||
let currentBatchNum = cmdStartBatch > 0 ? cmdStartBatch : 1;
|
||||
let totalBatches = 0; // Initialize totalBatches
|
||||
let totalDays = 0; // Initialize totalDays
|
||||
|
||||
console.log(`Starting snapshot backfill process...`);
|
||||
console.log(`SQL Function definition file: ${SQL_FUNCTION_FILE}`);
|
||||
if (!fs.existsSync(SQL_FUNCTION_FILE)) {
|
||||
console.error(`FATAL: SQL file not found at ${SQL_FUNCTION_FILE}`);
|
||||
process.exit(1); // Exit early if file doesn't exist
|
||||
}
|
||||
|
||||
try {
|
||||
// Set up a connection with higher memory limits
|
||||
connection = await getConnection({
|
||||
// Add performance-related settings
|
||||
application_name: 'backfill_snapshots',
|
||||
statement_timeout: PG_STATEMENT_TIMEOUT_MS, // 30 min timeout per statement
|
||||
// These parameters may need to be configured in your database:
|
||||
// work_mem: '1GB',
|
||||
// maintenance_work_mem: '2GB',
|
||||
// temp_buffers: '1GB',
|
||||
});
|
||||
|
||||
console.log('Database connection acquired.');
|
||||
|
||||
// --- Ensure Function Exists ---
|
||||
console.log('Ensuring database function is up-to-date...');
|
||||
try {
|
||||
const sqlFunctionDef = fs.readFileSync(SQL_FUNCTION_FILE, 'utf8');
|
||||
if (!sqlFunctionDef.includes('CREATE OR REPLACE FUNCTION backfill_daily_snapshots_range_final')) {
|
||||
throw new Error(`SQL file ${SQL_FUNCTION_FILE} does not seem to contain the function definition.`);
|
||||
}
|
||||
await connection.query(sqlFunctionDef); // Execute the whole file
|
||||
console.log('Database function `backfill_daily_snapshots_range_final` created/updated.');
|
||||
|
||||
// Add performance query hints to the database
|
||||
await connection.query(`
|
||||
-- Analyze tables for better query planning
|
||||
ANALYZE public.products;
|
||||
ANALYZE public.imported_daily_inventory;
|
||||
ANALYZE public.imported_product_stat_history;
|
||||
ANALYZE public.daily_product_snapshots;
|
||||
ANALYZE public.imported_product_current_prices;
|
||||
`).catch(err => {
|
||||
// Non-fatal if analyze fails
|
||||
console.warn('Failed to analyze tables (non-fatal):', err.message);
|
||||
});
|
||||
|
||||
} catch (err) {
|
||||
console.error(`Error processing SQL function file ${SQL_FUNCTION_FILE}:`, err);
|
||||
throw new Error(`Failed to create or replace DB function: ${err.message}`);
|
||||
}
|
||||
|
||||
// --- Prepare History Record ---
|
||||
console.log('Preparing calculation history record...');
|
||||
// Ensure history table exists (optional, could be done elsewhere)
|
||||
await connection.query(`
|
||||
CREATE TABLE IF NOT EXISTS public.calculate_history (
|
||||
id SERIAL PRIMARY KEY,
|
||||
start_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
end_time TIMESTAMPTZ,
|
||||
duration_seconds INTEGER,
|
||||
status VARCHAR(20) NOT NULL, -- e.g., 'running', 'completed', 'failed', 'cancelled'
|
||||
error_message TEXT,
|
||||
additional_info JSONB -- Store type, file, batch info etc.
|
||||
);
|
||||
`);
|
||||
// Mark previous runs of this type as potentially failed if they were left 'running'
|
||||
await connection.query(`
|
||||
UPDATE public.calculate_history
|
||||
SET status = 'failed', error_message = 'Interrupted by new run.'
|
||||
WHERE status = 'running' AND additional_info->>'type' = $1;
|
||||
`, [HISTORY_TYPE]);
|
||||
|
||||
// Create new history record
|
||||
const historyResult = await connection.query(`
|
||||
INSERT INTO public.calculate_history (start_time, status, additional_info)
|
||||
VALUES (NOW(), 'running', jsonb_build_object('type', $1::text, 'sql_file', $2::text, 'start_batch', $3::integer))
|
||||
RETURNING id;
|
||||
`, [HISTORY_TYPE, path.basename(SQL_FUNCTION_FILE), cmdStartBatch]);
|
||||
calculateHistoryId = historyResult.rows[0].id;
|
||||
console.log(`Calculation history record created with ID: ${calculateHistoryId}`);
|
||||
|
||||
|
||||
// --- Determine Date Range ---
|
||||
console.log('Determining date range...');
|
||||
let effectiveStartDate, effectiveEndDate;
|
||||
|
||||
// Use command-line dates if provided, otherwise query DB
|
||||
if (cmdStartDate) {
|
||||
effectiveStartDate = cmdStartDate;
|
||||
} else {
|
||||
const minDateResult = await connection.query(`
|
||||
SELECT LEAST(
|
||||
COALESCE((SELECT MIN(date) FROM public.imported_daily_inventory WHERE date > '1970-01-01'), CURRENT_DATE),
|
||||
COALESCE((SELECT MIN(date) FROM public.imported_product_stat_history WHERE date > '1970-01-01'), CURRENT_DATE)
|
||||
)::date as min_date;
|
||||
`);
|
||||
effectiveStartDate = minDateResult.rows[0]?.min_date || new Date().toISOString().split('T')[0]; // Fallback
|
||||
console.log(`Auto-detected start date: ${effectiveStartDate}`);
|
||||
}
|
||||
|
||||
if (cmdEndDate) {
|
||||
effectiveEndDate = cmdEndDate;
|
||||
} else {
|
||||
const maxDateResult = await connection.query(`
|
||||
SELECT GREATEST(
|
||||
COALESCE((SELECT MAX(date) FROM public.imported_daily_inventory WHERE date < CURRENT_DATE), '1970-01-01'::date),
|
||||
COALESCE((SELECT MAX(date) FROM public.imported_product_stat_history WHERE date < CURRENT_DATE), '1970-01-01'::date)
|
||||
)::date as max_date;
|
||||
`);
|
||||
// Ensure end date is not today or in the future
|
||||
effectiveEndDate = maxDateResult.rows[0]?.max_date || new Date(Date.now() - 86400000).toISOString().split('T')[0]; // Default yesterday
|
||||
if (new Date(effectiveEndDate) >= new Date(new Date().toISOString().split('T')[0])) {
|
||||
effectiveEndDate = new Date(Date.now() - 86400000).toISOString().split('T')[0]; // Set to yesterday if >= today
|
||||
}
|
||||
console.log(`Auto-detected end date: ${effectiveEndDate}`);
|
||||
}
|
||||
|
||||
// Validate dates
|
||||
const dStart = new Date(effectiveStartDate);
|
||||
const dEnd = new Date(effectiveEndDate);
|
||||
if (isNaN(dStart.getTime()) || isNaN(dEnd.getTime()) || dStart > dEnd) {
|
||||
throw new Error(`Invalid date range: Start "${effectiveStartDate}", End "${effectiveEndDate}"`);
|
||||
}
|
||||
|
||||
// --- Batch Processing ---
|
||||
totalDays = Math.ceil((dEnd - dStart) / (1000 * 60 * 60 * 24)) + 1; // Inclusive
|
||||
totalBatches = Math.ceil(totalDays / BATCH_SIZE_DAYS);
|
||||
|
||||
console.log(`Target Date Range: ${effectiveStartDate} to ${effectiveEndDate} (${totalDays} days)`);
|
||||
console.log(`Total Batches: ${totalBatches} (Batch Size: ${BATCH_SIZE_DAYS} days)`);
|
||||
console.log(`Starting from Batch: ${currentBatchNum}`);
|
||||
|
||||
// Initial progress update
|
||||
progress.outputProgress({
|
||||
status: 'running',
|
||||
operation: 'Starting Batch Processing',
|
||||
currentBatch: currentBatchNum,
|
||||
totalBatches: totalBatches,
|
||||
totalDays: totalDays,
|
||||
elapsed: '0s',
|
||||
remaining: 'Calculating...',
|
||||
rate: 0,
|
||||
historyId: calculateHistoryId // Include history ID in the object
|
||||
});
|
||||
|
||||
while (currentBatchNum <= totalBatches && !isCancelled) {
|
||||
const batchOffset = (currentBatchNum - 1) * BATCH_SIZE_DAYS;
|
||||
const batchStartDate = new Date(dStart);
|
||||
batchStartDate.setDate(dStart.getDate() + batchOffset);
|
||||
|
||||
const batchEndDate = new Date(batchStartDate);
|
||||
batchEndDate.setDate(batchStartDate.getDate() + BATCH_SIZE_DAYS - 1);
|
||||
|
||||
// Clamp batch end date to the overall effective end date
|
||||
if (batchEndDate > dEnd) {
|
||||
batchEndDate.setTime(dEnd.getTime());
|
||||
}
|
||||
|
||||
const batchStartDateStr = batchStartDate.toISOString().split('T')[0];
|
||||
const batchEndDateStr = batchEndDate.toISOString().split('T')[0];
|
||||
const batchStartTime = Date.now();
|
||||
|
||||
console.log(`\n--- Processing Batch ${currentBatchNum} / ${totalBatches} ---`);
|
||||
console.log(` Dates: ${batchStartDateStr} to ${batchEndDateStr}`);
|
||||
|
||||
// Execute the function for the batch
|
||||
try {
|
||||
progress.outputProgress({
|
||||
status: 'running',
|
||||
operation: `Executing DB function for batch ${currentBatchNum}...`,
|
||||
currentBatch: currentBatchNum,
|
||||
totalBatches: totalBatches,
|
||||
totalDays: totalDays,
|
||||
elapsed: progress.formatElapsedTime(overallStartTime),
|
||||
remaining: 'Executing...',
|
||||
rate: 0,
|
||||
historyId: calculateHistoryId
|
||||
});
|
||||
|
||||
// Performance improvement: Add batch processing hint
|
||||
await connection.query('SET LOCAL enable_parallel_append = on; SET LOCAL enable_parallel_hash = on; SET LOCAL max_parallel_workers_per_gather = 4;');
|
||||
|
||||
// Store promise in case we need to try and cancel (though not implemented forcefully)
|
||||
runningQueryPromise = connection.query(
|
||||
`SELECT backfill_daily_snapshots_range_final($1::date, $2::date);`,
|
||||
[batchStartDateStr, batchEndDateStr]
|
||||
);
|
||||
await runningQueryPromise; // Wait for the function call to complete
|
||||
runningQueryPromise = null; // Clear the promise
|
||||
|
||||
const batchDurationMs = Date.now() - batchStartTime;
|
||||
const daysInThisBatch = Math.ceil((batchEndDate - batchStartDate) / (1000 * 60 * 60 * 24)) + 1;
|
||||
processedDaysTotal += daysInThisBatch;
|
||||
|
||||
console.log(` Batch ${currentBatchNum} completed in ${progress.formatElapsedTime(batchStartTime)}.`);
|
||||
|
||||
// --- Update Progress & History ---
|
||||
const overallElapsedSec = Math.round((Date.now() - overallStartTime) / 1000);
|
||||
progress.outputProgress({
|
||||
status: 'running',
|
||||
operation: `Completed batch ${currentBatchNum}`,
|
||||
currentBatch: currentBatchNum,
|
||||
totalBatches: totalBatches,
|
||||
totalDays: totalDays,
|
||||
processedDays: processedDaysTotal,
|
||||
elapsed: progress.formatElapsedTime(overallStartTime),
|
||||
remaining: progress.estimateRemaining(overallStartTime, processedDaysTotal, totalDays),
|
||||
rate: progress.calculateRate(overallStartTime, processedDaysTotal),
|
||||
batchDuration: progress.formatElapsedTime(batchStartTime),
|
||||
historyId: calculateHistoryId
|
||||
});
|
||||
|
||||
// Save checkpoint in history
|
||||
await connection.query(`
|
||||
UPDATE public.calculate_history
|
||||
SET additional_info = jsonb_set(additional_info, '{last_completed_batch}', $1::jsonb)
|
||||
|| jsonb_build_object('last_processed_date', $2::text)
|
||||
WHERE id = $3::integer;
|
||||
`, [JSON.stringify(currentBatchNum), batchEndDateStr, calculateHistoryId]);
|
||||
|
||||
|
||||
} catch (batchError) {
|
||||
console.error(`\n--- ERROR in Batch ${currentBatchNum} (${batchStartDateStr} to ${batchEndDateStr}) ---`);
|
||||
console.error(' Database Error:', batchError.message);
|
||||
console.error(' DB Error Code:', batchError.code);
|
||||
// Log detailed error to history and re-throw to stop the process
|
||||
await connection.query(`
|
||||
UPDATE public.calculate_history
|
||||
SET status = 'failed',
|
||||
end_time = NOW(),
|
||||
duration_seconds = $1::integer,
|
||||
error_message = $2::text,
|
||||
additional_info = additional_info || jsonb_build_object('failed_batch', $3::integer, 'failed_date_range', $4::text)
|
||||
WHERE id = $5::integer;
|
||||
`, [
|
||||
Math.round((Date.now() - overallStartTime) / 1000),
|
||||
`Batch ${currentBatchNum} failed: ${batchError.message} (Code: ${batchError.code || 'N/A'})`,
|
||||
currentBatchNum,
|
||||
`${batchStartDateStr} to ${batchEndDateStr}`,
|
||||
calculateHistoryId
|
||||
]);
|
||||
throw batchError; // Stop execution
|
||||
}
|
||||
|
||||
currentBatchNum++;
|
||||
// Optional delay between batches
|
||||
// await new Promise(resolve => setTimeout(resolve, 500));
|
||||
|
||||
} // End while loop
|
||||
|
||||
// --- Final Outcome ---
|
||||
const finalStatus = isCancelled ? 'cancelled' : 'completed';
|
||||
const finalMessage = isCancelled ? `Calculation stopped after completing batch ${currentBatchNum - 1}.` : 'Historical snapshots backfill completed successfully.';
|
||||
const finalDurationSec = Math.round((Date.now() - overallStartTime) / 1000);
|
||||
|
||||
console.log(`\n--- Backfill ${finalStatus.toUpperCase()} ---`);
|
||||
console.log(finalMessage);
|
||||
console.log(`Total duration: ${progress.formatElapsedTime(overallStartTime)}`);
|
||||
|
||||
// Update history record
|
||||
await connection.query(`
|
||||
UPDATE public.calculate_history SET status = $1::calculation_status, end_time = NOW(), duration_seconds = $2::integer, error_message = $3
|
||||
WHERE id = $4::integer;
|
||||
`, [finalStatus, finalDurationSec, (isCancelled ? 'User cancelled' : null), calculateHistoryId]);
|
||||
|
||||
if (!isCancelled) {
|
||||
progress.clearProgress(); // Clear progress state only on successful completion
|
||||
} else {
|
||||
progress.outputProgress({ // Final cancelled status update
|
||||
status: 'cancelled',
|
||||
operation: finalMessage,
|
||||
currentBatch: currentBatchNum - 1,
|
||||
totalBatches: totalBatches,
|
||||
totalDays: totalDays,
|
||||
processedDays: processedDaysTotal,
|
||||
elapsed: progress.formatElapsedTime(overallStartTime),
|
||||
remaining: 'Cancelled',
|
||||
rate: 0,
|
||||
historyId: calculateHistoryId
|
||||
});
|
||||
}
|
||||
|
||||
return { success: true, status: finalStatus, message: finalMessage, duration: finalDurationSec };
|
||||
|
||||
} catch (error) {
|
||||
console.error('\n--- Backfill encountered an unrecoverable error ---');
|
||||
console.error(error.message);
|
||||
const finalDurationSec = Math.round((Date.now() - overallStartTime) / 1000);
|
||||
|
||||
// Update history if possible
|
||||
if (connection && calculateHistoryId) {
|
||||
try {
|
||||
await connection.query(`
|
||||
UPDATE public.calculate_history
|
||||
SET status = $1::calculation_status, end_time = NOW(), duration_seconds = $2::integer, error_message = $3::text
|
||||
WHERE id = $4::integer;
|
||||
`, [
|
||||
isCancelled ? 'cancelled' : 'failed',
|
||||
finalDurationSec,
|
||||
error.message,
|
||||
calculateHistoryId
|
||||
]);
|
||||
} catch (histError) {
|
||||
console.error("Failed to update history record with error state:", histError);
|
||||
}
|
||||
} else {
|
||||
console.error("Could not update history record (no ID or connection).");
|
||||
}
|
||||
|
||||
// FIX: Use initialized value or a default if loop never started
|
||||
const batchNumForError = currentBatchNum > cmdStartBatch ? currentBatchNum - 1 : cmdStartBatch - 1;
|
||||
|
||||
// Update progress.outputProgress call to match actual function signature
|
||||
try {
|
||||
// Create progress data object
|
||||
const progressData = {
|
||||
status: 'failed',
|
||||
operation: 'Backfill failed',
|
||||
message: error.message,
|
||||
currentBatch: batchNumForError,
|
||||
totalBatches: totalBatches,
|
||||
totalDays: totalDays,
|
||||
processedDays: processedDaysTotal,
|
||||
elapsed: progress.formatElapsedTime(overallStartTime),
|
||||
remaining: 'Failed',
|
||||
rate: 0,
|
||||
// Include history ID in progress data if needed
|
||||
historyId: calculateHistoryId
|
||||
};
|
||||
|
||||
// Call with single object parameter (not separate historyId)
|
||||
progress.outputProgress(progressData);
|
||||
} catch (progressError) {
|
||||
console.error('Failed to report progress:', progressError);
|
||||
}
|
||||
|
||||
return { success: false, status: 'failed', error: error.message, duration: finalDurationSec };
|
||||
|
||||
} finally {
|
||||
if (connection) {
|
||||
console.log('Releasing database connection.');
|
||||
connection.release();
|
||||
}
|
||||
// Close pool only if this script is meant to be standalone
|
||||
// If part of a larger app, the app should manage pool closure
|
||||
// console.log('Closing database pool.');
|
||||
// await closePool();
|
||||
}
|
||||
}
|
||||
|
||||
// --- Script Execution ---
|
||||
|
||||
// Parse command-line arguments
|
||||
const args = process.argv.slice(2);
|
||||
let cmdStartDateArg, cmdEndDateArg, cmdStartBatchArg = 1; // Default start batch is 1
|
||||
|
||||
for (let i = 0; i < args.length; i++) {
|
||||
if (args[i] === '--start-date' && args[i+1]) cmdStartDateArg = args[++i];
|
||||
else if (args[i] === '--end-date' && args[i+1]) cmdEndDateArg = args[++i];
|
||||
else if (args[i] === '--start-batch' && args[i+1]) cmdStartBatchArg = parseInt(args[++i], 10);
|
||||
}
|
||||
|
||||
if (isNaN(cmdStartBatchArg) || cmdStartBatchArg < 1) {
|
||||
console.warn(`Invalid --start-batch value. Defaulting to 1.`);
|
||||
cmdStartBatchArg = 1;
|
||||
}
|
||||
|
||||
// Run the backfill process
|
||||
backfillSnapshots(cmdStartDateArg, cmdEndDateArg, cmdStartBatchArg)
|
||||
.then(result => {
|
||||
if (result.success) {
|
||||
console.log(`\n✅ ${result.message} (Duration: ${result.duration}s)`);
|
||||
process.exitCode = 0; // Success
|
||||
} else {
|
||||
console.error(`\n❌ Backfill failed: ${result.error || 'Unknown error'} (Duration: ${result.duration}s)`);
|
||||
process.exitCode = 1; // Failure
|
||||
}
|
||||
})
|
||||
.catch(err => {
|
||||
console.error('\n❌ Unexpected error during backfill execution:', err);
|
||||
process.exitCode = 1; // Failure
|
||||
})
|
||||
.finally(async () => {
|
||||
// Ensure pool is closed if run standalone
|
||||
console.log('Backfill script finished. Closing pool.');
|
||||
await closePool(); // Make sure closePool exists and works in your db utils
|
||||
process.exit(process.exitCode); // Exit with appropriate code
|
||||
});
|
||||
161
inventory-server/old/backfill_historical_snapshots.sql
Normal file
161
inventory-server/old/backfill_historical_snapshots.sql
Normal file
@@ -0,0 +1,161 @@
|
||||
-- Description: Backfills the daily_product_snapshots table using imported historical unit data
|
||||
-- (daily inventory/stats) and historical price data (current prices table).
|
||||
-- - Uses imported daily sales/receipt UNIT counts for accuracy.
|
||||
-- - ESTIMATES historical stock levels using a forward calculation.
|
||||
-- - APPROXIMATES historical REVENUE using looked-up historical base prices.
|
||||
-- - APPROXIMATES historical COGS, PROFIT, and STOCK VALUE using CURRENT product costs/prices.
|
||||
-- Run ONCE after importing historical data and before initial product_metrics population.
|
||||
-- Dependencies: Core import tables (products), imported history tables (imported_daily_inventory,
|
||||
-- imported_product_stat_history, imported_product_current_prices),
|
||||
-- daily_product_snapshots table must exist.
|
||||
-- Frequency: Run ONCE.
|
||||
|
||||
CREATE OR REPLACE FUNCTION backfill_daily_snapshots_range_final(
|
||||
_start_date DATE,
|
||||
_end_date DATE
|
||||
)
|
||||
RETURNS VOID AS $$
|
||||
DECLARE
|
||||
_current_processing_date DATE := _start_date;
|
||||
_batch_start_time TIMESTAMPTZ;
|
||||
_row_count INTEGER;
|
||||
BEGIN
|
||||
RAISE NOTICE 'Starting FINAL historical snapshot backfill from % to %.', _start_date, _end_date;
|
||||
RAISE NOTICE 'Using historical units and historical prices (for revenue approximation).';
|
||||
RAISE NOTICE 'WARNING: Historical COGS, Profit, and Stock Value use CURRENT product costs/prices.';
|
||||
|
||||
-- Ensure end date is not in the future
|
||||
IF _end_date >= CURRENT_DATE THEN
|
||||
_end_date := CURRENT_DATE - INTERVAL '1 day';
|
||||
RAISE NOTICE 'Adjusted end date to % to avoid conflict with hourly script.', _end_date;
|
||||
END IF;
|
||||
|
||||
-- Performance: Create temporary table with product info to avoid repeated lookups
|
||||
CREATE TEMP TABLE IF NOT EXISTS temp_product_info AS
|
||||
SELECT
|
||||
pid,
|
||||
sku,
|
||||
COALESCE(landing_cost_price, cost_price, 0.00) as effective_cost_price,
|
||||
COALESCE(price, 0.00) as current_price,
|
||||
COALESCE(regular_price, 0.00) as current_regular_price
|
||||
FROM public.products;
|
||||
|
||||
-- Performance: Create index on temporary table
|
||||
CREATE INDEX IF NOT EXISTS temp_product_info_pid_idx ON temp_product_info(pid);
|
||||
|
||||
ANALYZE temp_product_info;
|
||||
|
||||
RAISE NOTICE 'Created temporary product info table with % products', (SELECT COUNT(*) FROM temp_product_info);
|
||||
|
||||
WHILE _current_processing_date <= _end_date LOOP
|
||||
_batch_start_time := clock_timestamp();
|
||||
RAISE NOTICE 'Processing date: %', _current_processing_date;
|
||||
|
||||
-- Get Daily Transaction Unit Info from imported history
|
||||
WITH DailyHistoryUnits AS (
|
||||
SELECT
|
||||
pids.pid,
|
||||
-- Prioritize daily_inventory, fallback to product_stat_history for sold qty
|
||||
COALESCE(di.amountsold, ps.qty_sold, 0)::integer as units_sold_today,
|
||||
COALESCE(di.qtyreceived, 0)::integer as units_received_today
|
||||
FROM
|
||||
(SELECT DISTINCT pid FROM temp_product_info) pids -- Ensure all products are considered
|
||||
LEFT JOIN public.imported_daily_inventory di
|
||||
ON pids.pid = di.pid AND di.date = _current_processing_date
|
||||
LEFT JOIN public.imported_product_stat_history ps
|
||||
ON pids.pid = ps.pid AND ps.date = _current_processing_date
|
||||
-- Removed WHERE clause to ensure snapshots are created even for days with 0 activity,
|
||||
-- allowing stock carry-over. The main query will handle products properly.
|
||||
),
|
||||
HistoricalPrice AS (
|
||||
-- Find the base price (qty_buy=1) active on the processing date
|
||||
SELECT DISTINCT ON (pid)
|
||||
pid,
|
||||
price_each
|
||||
FROM public.imported_product_current_prices
|
||||
WHERE
|
||||
qty_buy = 1
|
||||
-- Use TIMESTAMPTZ comparison logic:
|
||||
AND date_active <= (_current_processing_date + interval '1 day' - interval '1 second') -- Active sometime on or before end of processing day
|
||||
AND (date_deactive IS NULL OR date_deactive > _current_processing_date) -- Not deactivated before start of processing day
|
||||
-- Assuming 'active' flag isn't needed if dates are correct; add 'AND active != 0' if necessary
|
||||
ORDER BY
|
||||
pid, date_active DESC -- Get the most recently activated price
|
||||
),
|
||||
PreviousStock AS (
|
||||
-- Get the estimated stock from the PREVIOUS day snapshot
|
||||
SELECT pid, eod_stock_quantity
|
||||
FROM public.daily_product_snapshots
|
||||
WHERE snapshot_date = _current_processing_date - INTERVAL '1 day'
|
||||
)
|
||||
-- Insert into the daily snapshots table
|
||||
INSERT INTO public.daily_product_snapshots (
|
||||
snapshot_date, pid, sku,
|
||||
eod_stock_quantity, eod_stock_cost, eod_stock_retail, eod_stock_gross, stockout_flag,
|
||||
units_sold, units_returned,
|
||||
gross_revenue, discounts, returns_revenue,
|
||||
net_revenue, cogs, gross_regular_revenue, profit,
|
||||
units_received, cost_received,
|
||||
calculation_timestamp
|
||||
)
|
||||
SELECT
|
||||
_current_processing_date AS snapshot_date,
|
||||
p.pid,
|
||||
p.sku,
|
||||
-- Estimated EOD Stock (using historical daily units)
|
||||
-- Handle potential NULL from joins with COALESCE 0
|
||||
COALESCE(ps.eod_stock_quantity, 0) + COALESCE(dh.units_received_today, 0) - COALESCE(dh.units_sold_today, 0) AS estimated_eod_stock,
|
||||
-- Valued Stock (using estimated stock and CURRENT prices/costs - APPROXIMATION)
|
||||
GREATEST(0, COALESCE(ps.eod_stock_quantity, 0) + COALESCE(dh.units_received_today, 0) - COALESCE(dh.units_sold_today, 0)) * p.effective_cost_price AS eod_stock_cost,
|
||||
GREATEST(0, COALESCE(ps.eod_stock_quantity, 0) + COALESCE(dh.units_received_today, 0) - COALESCE(dh.units_sold_today, 0)) * p.current_price AS eod_stock_retail, -- Stock retail uses current price
|
||||
GREATEST(0, COALESCE(ps.eod_stock_quantity, 0) + COALESCE(dh.units_received_today, 0) - COALESCE(dh.units_sold_today, 0)) * p.current_regular_price AS eod_stock_gross, -- Stock gross uses current regular price
|
||||
-- Stockout Flag (based on estimated stock)
|
||||
(COALESCE(ps.eod_stock_quantity, 0) + COALESCE(dh.units_received_today, 0) - COALESCE(dh.units_sold_today, 0)) <= 0 AS stockout_flag,
|
||||
|
||||
-- Today's Unit Aggregates from History
|
||||
COALESCE(dh.units_sold_today, 0) as units_sold,
|
||||
0 AS units_returned, -- Placeholder: Cannot determine returns from daily summary
|
||||
|
||||
-- Monetary Values using looked-up Historical Price and CURRENT Cost/RegPrice
|
||||
COALESCE(dh.units_sold_today, 0) * COALESCE(hp.price_each, p.current_price) AS gross_revenue, -- Approx Revenue
|
||||
0 AS discounts, -- Placeholder
|
||||
0 AS returns_revenue, -- Placeholder
|
||||
COALESCE(dh.units_sold_today, 0) * COALESCE(hp.price_each, p.current_price) AS net_revenue, -- Approx Net Revenue
|
||||
COALESCE(dh.units_sold_today, 0) * p.effective_cost_price AS cogs, -- Approx COGS (uses CURRENT cost)
|
||||
COALESCE(dh.units_sold_today, 0) * p.current_regular_price AS gross_regular_revenue, -- Approx Gross Regular Revenue
|
||||
-- Approx Profit
|
||||
(COALESCE(dh.units_sold_today, 0) * COALESCE(hp.price_each, p.current_price)) - (COALESCE(dh.units_sold_today, 0) * p.effective_cost_price) AS profit,
|
||||
|
||||
COALESCE(dh.units_received_today, 0) as units_received,
|
||||
-- Estimate received cost using CURRENT product cost
|
||||
COALESCE(dh.units_received_today, 0) * p.effective_cost_price AS cost_received, -- Approx
|
||||
|
||||
clock_timestamp() -- Timestamp of this specific calculation
|
||||
FROM temp_product_info p -- Use the temp table for better performance
|
||||
LEFT JOIN PreviousStock ps ON p.pid = ps.pid
|
||||
LEFT JOIN DailyHistoryUnits dh ON p.pid = dh.pid -- Join today's historical activity
|
||||
LEFT JOIN HistoricalPrice hp ON p.pid = hp.pid -- Join the looked-up historical price
|
||||
-- Optimization: Only process products with activity or previous stock
|
||||
WHERE (dh.units_sold_today > 0 OR dh.units_received_today > 0 OR COALESCE(ps.eod_stock_quantity, 0) > 0)
|
||||
|
||||
ON CONFLICT (snapshot_date, pid) DO NOTHING; -- Avoid errors if rerunning parts, but prefer clean runs
|
||||
|
||||
GET DIAGNOSTICS _row_count = ROW_COUNT;
|
||||
RAISE NOTICE 'Processed %: Inserted/Skipped % rows. Duration: %',
|
||||
_current_processing_date,
|
||||
_row_count,
|
||||
clock_timestamp() - _batch_start_time;
|
||||
|
||||
_current_processing_date := _current_processing_date + INTERVAL '1 day';
|
||||
|
||||
END LOOP;
|
||||
|
||||
-- Clean up temporary tables
|
||||
DROP TABLE IF EXISTS temp_product_info;
|
||||
|
||||
RAISE NOTICE 'Finished FINAL historical snapshot backfill.';
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
-- Example usage:
|
||||
-- SELECT backfill_daily_snapshots_range_final('2023-01-01'::date, '2023-12-31'::date);
|
||||
Reference in New Issue
Block a user